Generate mock access token when access env

This commit is contained in:
Patrick Smith 2019-01-07 17:29:53 -05:00
parent c89e5b824c
commit 91419af71a
4 changed files with 85 additions and 3 deletions

View File

@ -25,6 +25,13 @@ class CloudProviderInterface:
"""
raise NotImplementedError()
def get_access_token(self, environment_role): # pragma: no cover
"""Takes an `atst.model.EnvironmentRole` object and returns a federated
access token that gives the specified user access to the specified
environment with the proper permissions.
"""
raise NotImplementedError()
class MockCloudProvider(CloudProviderInterface):
def create_application(self, name):
@ -39,3 +46,11 @@ class MockCloudProvider(CloudProviderInterface):
def delete_role(self, environment_role):
# Currently nothing to do.
pass
def get_access_token(self, environment_role):
# for now, just create a mock token using the user and environement
# cloud IDs and the name of the role in the environment
user_id = str(environment_role.user.id)
env_id = environment_role.environment.cloud_id or ""
role_details = environment_role.role
return "::".join([user_id, env_id, role_details])

View File

@ -1,6 +1,15 @@
from flask import render_template, request as http_request, g, redirect, url_for
from flask import (
current_app as app,
g,
redirect,
render_template,
request as http_request,
url_for,
)
from . import workspaces_bp
from atst.domain.environment_roles import EnvironmentRoles
from atst.domain.exceptions import UnauthorizedError
from atst.domain.projects import Projects
from atst.domain.workspaces import Workspaces
from atst.forms.project import NewProjectForm, ProjectForm
@ -76,3 +85,15 @@ def update_project(workspace_id, project_id):
project=project,
form=form,
)
@workspaces_bp.route("/workspaces/<workspace_id>/environments/<environment_id>/access")
def access_environment(workspace_id, environment_id):
env_role = EnvironmentRoles.get(g.current_user.id, environment_id)
if not env_role:
raise UnauthorizedError(
g.current_user, "access environment {}".format(environment_id)
)
else:
token = app.csp.cloud.get_access_token(env_role)
return redirect(url_for("atst.csp_environment_access", token=token))

View File

@ -34,7 +34,7 @@
<ul>
{% for environment in project.environments %}
<li class='block-list__item project-list-item__environment'>
<a href='{{ url_for("atst.csp_environment_access")}}' target='_blank' rel='noopener noreferrer' class='project-list-item__environment__link'>
<a href='{{ url_for("workspaces.access_environment", workspace_id=workspace.id, environment_id=environment.id)}}' target='_blank' rel='noopener noreferrer' class='project-list-item__environment__link'>
{{ Icon('link') }}
<span>{{ environment.name }}</span>
</a>

View File

@ -1,6 +1,13 @@
from flask import url_for
from tests.factories import UserFactory, WorkspaceFactory
from tests.factories import (
UserFactory,
WorkspaceFactory,
WorkspaceRoleFactory,
EnvironmentRoleFactory,
EnvironmentFactory,
ProjectFactory,
)
from atst.domain.projects import Projects
from atst.domain.workspaces import Workspaces
from atst.models.workspace_role import Status as WorkspaceRoleStatus
@ -125,3 +132,42 @@ def test_user_without_permission_cannot_update_project(client, user_session):
assert response.status_code == 404
assert project.name == "Great Project"
assert project.description == "Cool stuff happening here!"
def create_environment(user):
workspace = WorkspaceFactory.create()
workspace_role = WorkspaceRoleFactory.create(workspace=workspace, user=user)
project = ProjectFactory.create(workspace=workspace)
return EnvironmentFactory.create(project=project, name="new environment!")
def test_environment_access_with_env_role(client, user_session):
user = UserFactory.create()
environment = create_environment(user)
env_role = EnvironmentRoleFactory.create(
user=user, environment=environment, role="developer"
)
user_session(user)
response = client.get(
url_for(
"workspaces.access_environment",
workspace_id=environment.workspace.id,
environment_id=environment.id,
)
)
assert response.status_code == 302
assert "csp-environment-access" in response.location
def test_environment_access_with_no_role(client, user_session):
user = UserFactory.create()
environment = create_environment(user)
user_session(user)
response = client.get(
url_for(
"workspaces.access_environment",
workspace_id=environment.workspace.id,
environment_id=environment.id,
)
)
assert response.status_code == 404