darkhelm/notifications-admin
1
0
Fork 0
mirror of https://github.com/GSA/notifications-admin.git synced 2026-07-18 05:30:21 -04:00
Code Issues Packages Projects Releases Wiki Activity

cleanup

Browse Source
This commit is contained in:
Kenneth Kehl
2025-09-09 12:43:15 -07:00
parent 3d7aee19d2
commit 5d83cc43b2
4 changed files with 43 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files

32
app/main/views/register.py
View File

@@ -164,7 +164,7 @@ def set_up_your_profile():
abort(403, "Login.gov state not detected #invites")
state_key = f"login-state-{unquote(state)}"
current_app.logger.debug(hilite(f"Register tries to fetch state_key {state_key}"))
debug_msg(f"Register tries to fetch state_key {state_key}")
stored_state = unquote(redis_client.get(state_key).decode("utf8"))
if state != stored_state:
@@ -189,18 +189,10 @@ def set_up_your_profile():
invite_data = redis_client.get(f"invitedata-{state}")
# TODO fails here.
invite_data = json.loads(invite_data)
current_app.logger.debug(hilite(f"HERE IS INVITE DATA {invite_data}"))
is_org_invite = False
if invite_data.get("invited_user_id"):
invited_user_id = invite_data["invited_user_id"]
invited_user_email_address = get_invited_user_email_address(invited_user_id)
else:
invited_user_id = invite_data["id"]
is_org_invite = True
invited_user_email_address = get_invited_org_user_email_address(
invited_user_id
)
is_org_invite, invited_user_id, invited_user_email_address = (
process_invited_user(invite_data)
)
current_app.logger.info(
f"#invites: does user email match expected? {user_email == invited_user_email_address}"
@@ -326,8 +318,6 @@ def invited_user_accept_invite(invited_user_id):
def invited_org_user_accept_invite(invited_user_id):
invited_user = InvitedOrgUser.by_id(invited_user_id)
current_app.logger.debug(hilite(f"INVITED ORG USER {invited_user.serialize()}"))
if invited_user.status == InvitedUserStatus.EXPIRED:
current_app.logger.error("User invitation has expired")
flash(
@@ -347,3 +337,17 @@ def invited_org_user_accept_invite(invited_user_id):
def debug_msg(msg):
current_app.logger.debug(hilite(msg))
def process_invited_user(invite_data):
is_org_invite = False
if invite_data.get("invited_user_id"):
invited_user_id = invite_data["invited_user_id"]
invited_user_email_address = get_invited_user_email_address(invited_user_id)
else:
invited_user_id = invite_data["id"]
is_org_invite = True
invited_user_email_address = get_invited_org_user_email_address(invited_user_id)
return is_org_invite, invited_user_id, invited_user_email_address

31
app/models/user.py
View File

@@ -202,7 +202,6 @@ class User(JSONModel, UserMixin):
@property
def is_gov_user(self):
is_gov = is_gov_user(self.email_address)
# current_app.logger.info(f"User {self.id} is_gov_user: {is_gov}")
return is_gov
@property
@@ -211,9 +210,6 @@ class User(JSONModel, UserMixin):
@property
def platform_admin(self):
# current_app.logger.warning(
# f"Checking User {self.id} for platform admin: {self._platform_admin}"
# )
return self._platform_admin and not session.get(
"disable_platform_admin_view", False
)
@@ -243,27 +239,17 @@ class User(JSONModel, UserMixin):
# we shouldn't have any pages that require permissions, but don't specify a service or organization.
# use @user_is_platform_admin for platform admin only pages
# raise NotImplementedError
# current_app.logger.warning(f"VIEW ARGS ARE {request.view_args}")
pass
# platform admins should be able to do most things (except eg send messages, or create api keys)
if self.platform_admin and not restrict_admin_usage:
current_app.logger.debug(
"has_permissions is true because user is platform_admin"
)
return True
if org_id:
value = self.belongs_to_organization(org_id)
if not value:
# TODO this is sketch! Fix this!
# This is temporary to restore org invite functionality only
# self.add_to_organization(org_id)
value = self.belongs_to_organization(org_id)
current_app.logger.debug(
f"has_permissions returns org: {org_id} returning {value}"
)
current_app.logger.debug(
f"has_permissions returns org: {org_id} returning {value}"
)
return value
if not permissions and self.belongs_to_service(service_id):
@@ -323,11 +309,6 @@ class User(JSONModel, UserMixin):
abort(403)
def belongs_to_organization(self, organization_id):
# TODO this is sketch! Fix this!
# This is temporary to restore org invite functionality only
# if str(organization_id) not in self.organization_ids:
# self.add_to_organization(organization_id)
return str(organization_id) in self.organization_ids
@property
@@ -575,17 +556,11 @@ class InvitedUser(JSONModel):
return cls.by_id(invited_user_id) if invited_user_id else None
def has_permissions(self, *permissions):
# current_app.logger.warning(
# f"Checking invited user {self.id} for permissions: {permissions}"
# )
if self.status == InvitedUserStatus.CANCELLED:
return False
return set(self.permissions) > set(permissions)
def has_permission_for_service(self, service_id, permission):
# current_app.logger.warn(
# f"Checking invited user {self.id} for permission: {permission} on service {service_id}"
# )
if self.status == InvitedUserStatus.CANCELLED:
return False
return self.service == service_id and permission in self.permissions

11
app/notify_client/org_invite_api_client.py
View File

@@ -7,7 +7,6 @@ from flask import current_app, request
from app import redis_client
from app.enums import InvitedOrgUserStatus
from app.notify_client import NotifyAdminAPIClient, _attach_current_user
from app.utils import hilite
from notifications_utils.url_safe_token import generate_token
@@ -34,23 +33,16 @@ class OrgInviteApiClient(NotifyAdminAPIClient):
)
state_key = f"login-state-{unquote(state)}"
redis_client.set(state_key, state, ex=ttl)
current_app.logger.debug(
hilite(f"SET THE STATE KEY TO {state} with state_key {state_key}")
)
# make and store the nonce
nonce = secrets.token_urlsafe()
nonce_key = f"login-nonce-{unquote(nonce)}"
redis_client.set(nonce_key, nonce, ex=ttl) # save the nonce to redis.
current_app.logger.debug(
hilite(f"SET THE STATE KEY TO {state} with state_key {state_key}")
)
data["nonce"] = nonce # This is passed to api for the invite url.
data["state"] = state # This is passed to api for the invite url.
resp = self.post(url="/organization/{}/invite".format(org_id), data=data)
current_app.logger.debug(hilite(f"RESP is {resp}"))
invite_data_key = f"invitedata-{unquote(state)}"
# For historical reasons 'invite' signifies a service invite
@@ -61,9 +53,6 @@ class OrgInviteApiClient(NotifyAdminAPIClient):
redis_invite_data = resp["data"]
redis_invite_data = json.dumps(redis_invite_data)
redis_client.set(invite_data_key, redis_invite_data, ex=ttl)
current_app.logger.debug(
hilite(f"SET invite_data_key {invite_data_key} to {redis_invite_data}")
)
return resp["data"]

23
tests/app/main/views/test_register.py
View File

@@ -6,7 +6,7 @@ import pytest
from flask import url_for
from app.enums import ServicePermission
from app.main.views.register import check_invited_user_email_address_matches_expected
from app.main.views.register import check_invited_user_email_address_matches_expected, process_invited_user
from app.models.user import User
@@ -446,3 +446,24 @@ def test_decode_state(encoded_invite_data):
"permissions": ["manage_everything"],
"service_id": "service",
}
#@pytest.mark.parametrize(
# ("user_services", "user_organizations", "expected_status", "organization_checked"),
# [
# ([SERVICE_ONE_ID], [], 200, False),
@pytest.mark.parametrize(
("invite_data", "expected_is_org_invite", "expected_user_id"),
[
({"invited_user_id": "service_p"}, False, "service_p"),
({"id": "org_p"}, True, "org_p"),
]
)
def test_process_invited_user(invite_data, expected_is_org_invite, expected_user_id, mocker):
mocker.patch("app.main.views.register.get_invited_user_email_address")
mocker.patch("app.main.views.register.get_invited_org_user_email_address")
is_org_invite, invited_user_id, _ = process_invited_user(invite_data)
assert expected_is_org_invite == is_org_invite
assert expected_user_id == invited_user_id