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

Merge pull request #2901 from GSA/org_invites

Browse Source 
fix org invites
This commit is contained in:
ccostino
2025-09-10 17:24:31 -04:00
committed by GitHub
parent f1b87f56ca 4a14bf6f89
commit 524e8c87e8
6 changed files with 165 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
.ds.baseline
View File

@@ -527,7 +527,7 @@
"filename": "tests/app/main/views/test_register.py",
"hashed_secret": "bdbb156d25d02fd7792865824201dda1c60f4473",
"is_verified": false,
"line_number": 115,
"line_number": 118,
"is_secret": false
},
{
@@ -535,7 +535,7 @@
"filename": "tests/app/main/views/test_register.py",
"hashed_secret": "5baa61e4c9b93f3f0682250b6cf8331b7ee68fd8",
"is_verified": false,
"line_number": 185,
"line_number": 188,
"is_secret": false
},
{
@@ -543,7 +543,7 @@
"filename": "tests/app/main/views/test_register.py",
"hashed_secret": "bb5b7caa27d005d38039e3797c3ddb9bcd22c3c8",
"is_verified": false,
"line_number": 256,
"line_number": 259,
"is_secret": false
}
],
@@ -634,5 +634,5 @@
}
]
},
"generated_at": "2025-08-21T16:06:57Z"
"generated_at": "2025-09-09T19:57:05Z"
}

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

@@ -164,7 +164,9 @@ def set_up_your_profile():
abort(403, "Login.gov state not detected #invites")
state_key = f"login-state-{unquote(state)}"
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:
flash("Internal error: cannot recognize stored state")
abort(403, "Internal error: cannot recognize stored state #invites")
@@ -185,21 +187,33 @@ def set_up_your_profile():
f"#invites: Got the user_email and user_uuid {user_uuid} from login.gov"
)
invite_data = redis_client.get(f"invitedata-{state}")
# TODO fails here.
invite_data = json.loads(invite_data)
invited_user_id = invite_data["invited_user_id"]
invited_user_email_address = get_invited_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}"
)
check_invited_user_email_address_matches_expected(
user_email, invited_user_email_address
)
if is_org_invite:
invited_org_user_accept_invite(invited_user_id)
current_app.logger.info(
f"#invites: accepted org invite user with invited_user_id \
{invited_user_id}"
)
else:
invited_user_accept_invite(invited_user_id)
current_app.logger.info(
f"#invites: accepted invite user with invited_user_id \
{invited_user_id}"
)
invited_user_accept_invite(invited_user_id)
current_app.logger.info(
f"#invites: accepted invite user with invited_user_id \
{invited_user_id} to service {invite_data['service_id']}"
)
# We need to avoid taking a second trip through the login.gov code because we cannot pull the
# access token twice. So once we retrieve these values, let's park them in redis for 15 minutes
put_invite_data_in_redis(
@@ -239,22 +253,31 @@ def set_up_your_profile():
current_app.logger.info("#invites: going to activate user")
activate_user(user["id"])
current_app.logger.info(f"#invites: activated user with user.id {user['id']}")
usr = User.from_id(user["id"])
usr.add_to_service(
invite_data["service_id"],
invite_data["permissions"],
invite_data["folder_permissions"],
invite_data["from_user_id"],
)
if invite_data.get("service_id"):
usr = User.from_id(user["id"])
# notify-admin-1766
# redirect new users to templates area of new service instead of dashboard
service_id = invite_data["service_id"]
url = url_for(".service_dashboard", service_id=service_id)
url = f"{url}/templates"
current_app.logger.info(f"#invites redirecting to {url}")
return redirect(url)
usr.add_to_service(
invite_data["service_id"],
invite_data["permissions"],
invite_data["folder_permissions"],
invite_data["from_user_id"],
)
# notify-admin-1766
# redirect new users to templates area of new service instead of dashboard
service_id = invite_data["service_id"]
url = url_for(".service_dashboard", service_id=service_id)
url = f"{url}/templates"
current_app.logger.info(f"#invites redirecting to {url}")
return redirect(url)
else:
usr = User.from_id(user["id"])
org_id = invite_data["organization"]
usr.add_to_organization(org_id)
url = url_for(".organization_dashboard", org_id=org_id)
current_app.logger.info(f"#invites redirecting to {url}")
return redirect(url)
# we take two trips through this method, but should only hit this
# line on the first trip. On the second trip, we should get redirected
@@ -269,25 +292,67 @@ def get_invited_user_email_address(invited_user_id):
return invited_user.email_address
def get_invited_org_user_email_address(invited_user_id):
# InvitedUser is an unhashable type and hard to mock in tests
# so this convenience method is a workaround for that
invited_user = InvitedOrgUser.by_id(invited_user_id)
return invited_user.email_address
def invited_user_accept_invite(invited_user_id):
invited_user = InvitedUser.by_id(invited_user_id)
if invited_user.status == InvitedUserStatus.EXPIRED:
current_app.logger.error("User invitation has expired")
current_app.logger.error("Service invitation has expired")
flash(
"Your invitation has expired; please contact the person who invited you for additional help."
"Your service invitation has expired; please contact the person who invited you for additional help."
)
abort(401, "Your invitation has expired #invites")
abort(401, "Your service invitation has expired #invites")
if invited_user.status == InvitedUserStatus.CANCELLED:
current_app.logger.error("User invitation has been cancelled")
current_app.logger.error("Service invitation has been cancelled")
flash(
"Your invitation is no longer valid; please contact the person who invited you for additional help."
"Your service invitation is no longer valid; please contact the person who invited you for additional help."
)
abort(401, "Your invitation was canceled #invites")
abort(401, "Your service invitation was canceled #invites")
invited_user.accept_invite()
def invited_org_user_accept_invite(invited_user_id):
invited_user = InvitedOrgUser.by_id(invited_user_id)
if invited_user.status == InvitedUserStatus.EXPIRED:
current_app.logger.error("Organization invitation has expired")
flash(
"Your organization invitation has expired; please contact the person who invited you for additional help."
)
abort(401, "Your organization invitation has expired #invites")
if invited_user.status == InvitedUserStatus.CANCELLED:
current_app.logger.error("Organization invitation has been cancelled")
flash(
"Your organization invitation is no longer valid; \
please contact the person who invited you for additional help."
)
abort(401, "Your organization invitation was canceled #invites")
invited_user.accept_invite()
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

14
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,14 +239,10 @@ 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:
@@ -564,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

38
app/notify_client/org_invite_api_client.py
View File

@@ -1,5 +1,13 @@
import json
import secrets
from urllib.parse import unquote
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 notifications_utils.url_safe_token import generate_token
class OrgInviteApiClient(NotifyAdminAPIClient):
@@ -15,7 +23,37 @@ class OrgInviteApiClient(NotifyAdminAPIClient):
"invite_link_host": self.admin_url,
}
data = _attach_current_user(data)
ttl = 24 * 60 * 60
# make and store the state
state = generate_token(
str(request.remote_addr),
current_app.config["SECRET_KEY"],
current_app.config["DANGEROUS_SALT"],
)
state_key = f"login-state-{unquote(state)}"
redis_client.set(state_key, state, ex=ttl)
# 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.
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)
invite_data_key = f"invitedata-{unquote(state)}"
# For historical reasons 'invite' signifies a service invite
# and 'data' signifies an org invite
if resp.get("invite"):
redis_invite_data = resp["invite"]
else:
redis_invite_data = resp["data"]
redis_invite_data = json.dumps(redis_invite_data)
redis_client.set(invite_data_key, redis_invite_data, ex=ttl)
return resp["data"]
def get_invites_for_organization(self, org_id):

2
tests/app/main/test_permissions.py
View File

@@ -81,6 +81,8 @@ def test_services_pages_that_org_users_are_allowed_to_see(
endpoint,
service_id=SERVICE_ONE_ID,
_expected_status=expected_status,
nonce="dummy-nonce",
state="dummy-state",
)
assert mock_get_service.called is organization_checked

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

@@ -6,7 +6,10 @@ 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 +449,28 @@ 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