jasquat c056b89006 Feature/group mapping from keycloak (#457)
* some basics to set a user groups based on info in keycloak w/ burnettk

* test for adding groups from token now passes

* do not remove users from groups when running refresh_permissions if specified

---------

Co-authored-by: jasquat <jasquat@users.noreply.github.com>
2023-09-07 11:17:40 -04:00

88 lines
3.6 KiB
Python

import ast
import base64
import time
from flask.app import Flask
from flask.testing import FlaskClient
from spiffworkflow_backend.models.db import db
from spiffworkflow_backend.models.user import UserModel
from spiffworkflow_backend.services.authentication_service import AuthenticationService
from spiffworkflow_backend.services.authorization_service import AuthorizationService
from spiffworkflow_backend.services.authorization_service import GroupPermissionsDict
from tests.spiffworkflow_backend.helpers.base_test import BaseTest
class TestAuthentication(BaseTest):
def test_get_login_state(self) -> None:
redirect_url = "http://example.com/"
state = AuthenticationService.generate_state(redirect_url)
state_dict = ast.literal_eval(base64.b64decode(state).decode("utf-8"))
assert isinstance(state_dict, dict)
assert "redirect_url" in state_dict.keys()
assert state_dict["redirect_url"] == redirect_url
def test_properly_adds_user_to_groups_from_token_on_login(
self,
app: Flask,
client: FlaskClient,
with_db_and_bpmn_file_cleanup: None,
) -> None:
user = self.find_or_create_user("testing@e.com")
user.email = "testing@e.com"
user.service = app.config["SPIFFWORKFLOW_BACKEND_OPEN_ID_SERVER_URL"]
db.session.add(user)
db.session.commit()
access_token = user.encode_auth_token(
{
"groups": ["group_one", "group_two"],
"iss": app.config["SPIFFWORKFLOW_BACKEND_OPEN_ID_SERVER_URL"],
"aud": "spiffworkflow-backend",
"iat": round(time.time()),
"exp": round(time.time()) + 1000,
}
)
response = client.post(
f"/v1.0/login_with_access_token?access_token={access_token}",
)
assert response.status_code == 200
assert len(user.groups) == 3
group_identifiers = [g.identifier for g in user.groups]
assert sorted(group_identifiers) == ["everybody", "group_one", "group_two"]
access_token = user.encode_auth_token(
{
"groups": ["group_one"],
"iss": app.config["SPIFFWORKFLOW_BACKEND_OPEN_ID_SERVER_URL"],
"aud": "spiffworkflow-backend",
"iat": round(time.time()),
"exp": round(time.time()) + 1000,
}
)
response = client.post(
f"/v1.0/login_with_access_token?access_token={access_token}",
)
assert response.status_code == 200
user = UserModel.query.filter_by(username=user.username).first()
assert len(user.groups) == 2
group_identifiers = [g.identifier for g in user.groups]
assert sorted(group_identifiers) == ["everybody", "group_one"]
# make sure running refresh_permissions doesn't remove the user from the group
group_info: list[GroupPermissionsDict] = [
{
"users": [],
"name": "group_one",
"permissions": [{"actions": ["create", "read"], "uri": "PG:hey"}],
}
]
AuthorizationService.refresh_permissions(group_info, group_permissions_only=True)
user = UserModel.query.filter_by(username=user.username).first()
assert len(user.groups) == 2
group_identifiers = [g.identifier for g in user.groups]
assert sorted(group_identifiers) == ["everybody", "group_one"]
self.assert_user_has_permission(user, "read", "/v1.0/process-groups/hey")
self.assert_user_has_permission(user, "read", "/v1.0/process-groups/hey:yo")