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

Merge pull request #1282 from GSA/notify-admin-1277

Browse Source 
update login.gov stuff to use user uuid instead of email (notify-admi…
This commit is contained in:
Carlo Costino
2024-03-11 11:55:04 -04:00
committed by GitHub
parent b489f49179 492f798274
commit bdea60adb9
2 changed files with 15 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
app/main/views/sign_in.py
View File

@@ -53,22 +53,19 @@ def _get_access_token(code, state):
# JWT expiration time (10 minute maximum) # JWT expiration time (10 minute maximum)
"exp": int(time.time()) + (10 * 60), "exp": int(time.time()) + (10 * 60),
} }
current_app.logger.warning(f"Here is the raw payload {payload}")
token = jwt.encode(payload, keystring, algorithm="RS256") token = jwt.encode(payload, keystring, algorithm="RS256")
base_url = f"{access_token_url}?" base_url = f"{access_token_url}?"
cli_assert = f"client_assertion={token}" cli_assert = f"client_assertion={token}"
cli_assert_type = "client_assertion_type=urn%3Aietf%3Aparams%3Aoauth%3Aclient-assertion-type%3Ajwt-bearer" cli_assert_type = "client_assertion_type=urn%3Aietf%3Aparams%3Aoauth%3Aclient-assertion-type%3Ajwt-bearer"
code_param = f"code={code}" code_param = f"code={code}"
url = f"{base_url}{cli_assert}&{cli_assert_type}&{code_param}&grant_type=authorization_code" url = f"{base_url}{cli_assert}&{cli_assert_type}&{code_param}&grant_type=authorization_code"
current_app.logger.info(f"This is the url we use to get the access token: {url}")
headers = {"Authorization": "Bearer %s" % token} headers = {"Authorization": "Bearer %s" % token}
response = requests.post(url, headers=headers) response = requests.post(url, headers=headers)
current_app.logger.info(f"GOT A RESPONSE {response.json()}")
access_token = response.json()["access_token"] access_token = response.json()["access_token"]
return access_token return access_token
def _get_user_email(access_token): def _get_user_email_and_uuid(access_token):
headers = {"Authorization": "Bearer %s" % access_token} headers = {"Authorization": "Bearer %s" % access_token}
user_info_url = os.getenv("LOGIN_DOT_GOV_USER_INFO_URL") user_info_url = os.getenv("LOGIN_DOT_GOV_USER_INFO_URL")
user_attributes = requests.get( user_attributes = requests.get(
@@ -76,7 +73,8 @@ def _get_user_email(access_token):
headers=headers, headers=headers,
) )
user_email = user_attributes.json()["email"] user_email = user_attributes.json()["email"]
return user_email user_uuid = user_attributes.json()["sub"]
return user_email, user_uuid
@main.route("/sign-in", methods=(["GET", "POST"])) @main.route("/sign-in", methods=(["GET", "POST"]))
@@ -88,11 +86,11 @@ def sign_in():
login_gov_error = request.args.get("error") login_gov_error = request.args.get("error")
if code and state: if code and state:
access_token = _get_access_token(code, state) access_token = _get_access_token(code, state)
user_email = _get_user_email(access_token) user_email, user_uuid = _get_user_email_and_uuid(access_token)
redirect_url = request.args.get("next") redirect_url = request.args.get("next")
# activate the user # activate the user
user = user_api_client.get_user_by_email(user_email) user = user_api_client.get_user_by_uuid_or_email(user_uuid, user_email)
activate_user(user["id"]) activate_user(user["id"])
return redirect(url_for("main.show_accounts_or_dashboard", next=redirect_url)) return redirect(url_for("main.show_accounts_or_dashboard", next=redirect_url))

10
app/notify_client/user_api_client.py
View File

@@ -44,6 +44,16 @@ class UserApiClient(NotifyAdminAPIClient):
user_data = self.post("/user/email", data={"email": email_address}) user_data = self.post("/user/email", data={"email": email_address})
return user_data["data"] return user_data["data"]
def get_user_by_uuid_or_email(self, user_uuid, email_address):
user_data = self.post(
"/user/get-login-gov-user",
data={"login_uuid": user_uuid, "email": email_address},
)
if user_data is None:
raise Exception("User not found")
return user_data["data"]
def get_user_by_email_or_none(self, email_address): def get_user_by_email_or_none(self, email_address):
try: try:
return self.get_user_by_email(email_address) return self.get_user_by_email(email_address)