Merge pull request #2024 from GSA/USN-COMPLY-50-Verify_Nonce_for_Invite

US-NOTIFY-COMPLY 50: Verify Nonce For Invite

app/main/views/index.py

@@ -1,18 +1,11 @@
 import os
 import secrets
+from urllib.parse import unquote
 
-from flask import (
+from flask import abort, current_app, redirect, render_template, request, url_for
-    abort,
-    current_app,
-    redirect,
-    render_template,
-    request,
-    session,
-    url_for,
-)
 from flask_login import current_user
 
-from app import status_api_client
+from app import redis_client, status_api_client
 from app.formatters import apply_html_class, convert_markdown_template
 from app.main import main
 from app.main.views.pricing import CURRENT_SMS_RATE
@@ -25,6 +18,7 @@ from notifications_utils.url_safe_token import generate_token
 def index():
     if current_user and current_user.is_authenticated:
         return redirect(url_for("main.choose_account"))
+
     token = generate_token(
         str(request.remote_addr),
         current_app.config["SECRET_KEY"],
@@ -33,8 +27,12 @@ def index():
     url = os.getenv("LOGIN_DOT_GOV_INITIAL_SIGNIN_URL")
     # handle unit tests
 
+    current_app.logger.warning(f"############### {str(request.remote_addr)}")
+
     nonce = secrets.token_urlsafe()
-    session["nonce"] = nonce
+
+    redis_key = f"login-nonce-{unquote(nonce)}"
+    redis_client.set(redis_key, nonce)
 
     if url is not None:
         url = url.replace("NONCE", nonce)

app/main/views/register.py

@@ -165,6 +165,7 @@ def set_up_your_profile():
 
     if redis_client.get(f"invitedata-{state}") is None:
         access_token = sign_in._get_access_token(code, state)
+
         debug_msg("Got the access token for login.gov")
         user_email, user_uuid = sign_in._get_user_email_and_uuid(access_token)
         debug_msg(

app/main/views/sign_in.py

@@ -1,8 +1,8 @@
-# import json
 import os
 import secrets
 import time
 import uuid
+from urllib.parse import unquote
 
 import jwt
 import requests
@@ -14,18 +14,17 @@ from flask import (
     redirect,
     render_template,
     request,
-    session,
     url_for,
 )
 from flask_login import current_user
 
-from app import login_manager, user_api_client
+from app import login_manager, redis_client, user_api_client
 from app.main import main
 from app.main.views.index import error
 from app.main.views.verify import activate_user
 from app.models.user import User
 from app.utils import hide_from_search_engines
-from app.utils.login import is_safe_redirect_url
+from app.utils.login import get_id_token, is_safe_redirect_url
 from app.utils.time import is_less_than_days_ago
 from app.utils.user import is_gov_user
 from notifications_utils.url_safe_token import generate_token
@@ -43,7 +42,6 @@ def _reformat_keystring(orig):  # pragma: no cover
 def _get_access_token(code, state):  # pragma: no cover
     client_id = os.getenv("LOGIN_DOT_GOV_CLIENT_ID")
     access_token_url = os.getenv("LOGIN_DOT_GOV_ACCESS_TOKEN_URL")
-    # certs_url = os.getenv("LOGIN_DOT_GOV_CERTS_URL")
     keystring = os.getenv("LOGIN_PEM")
     if " " in keystring:
         keystring = _reformat_keystring(keystring)
@@ -66,38 +64,14 @@ def _get_access_token(code, state):  # pragma: no cover
     response = requests.post(url, headers=headers)
 
     response_json = response.json()
+    id_token = get_id_token(response_json)
+    nonce = id_token["nonce"]
+    redis_key = f"login-nonce-{unquote(nonce)}"
+    stored_nonce = redis_client.get(redis_key).decode("utf8")
 
-    # TODO nonce check intermittently fails, investifix
+    if nonce != stored_nonce:
-    # Presumably the nonce is not yet in the session when there
+        current_app.logger.error(f"Nonce Error: {nonce} != {stored_nonce}")
-    # is an invite involved?
+        abort(403)
-
-    # try:
-    #     encoded_id_token = response_json["id_token"]
-    # except KeyError as e:
-    #     current_app.logger.exception(f"Error when getting id token {response_json}")
-    #     raise KeyError(f"'access_token' {response.json()}") from e
-
-    # Getting Login.gov signing keys for unpacking the id_token correctly.
-    # jwks = requests.get(certs_url).json()
-    # public_keys = {
-    #     jwk["kid"]: {
-    #         "key": jwt.algorithms.RSAAlgorithm.from_jwk(json.dumps(jwk)),
-    #         "algo": jwk["alg"],
-    #     }
-    #     for jwk in jwks["keys"]
-    # }
-    # kid = jwt.get_unverified_header(encoded_id_token)["kid"]
-    # pub_key = public_keys[kid]["key"]
-    # algo = public_keys[kid]["algo"]
-    # id_token = jwt.decode(
-    #    encoded_id_token, pub_key, audience=client_id, algorithms=[algo]
-    # )
-    # nonce = id_token["nonce"]
-
-    # saved_nonce = session.pop("nonce")
-    # if nonce != saved_nonce:
-    #     current_app.logger.error(f"Nonce Error: {nonce} != {saved_nonce}")
-    #     abort(403)
 
     try:
         access_token = response_json["access_token"]
@@ -237,7 +211,8 @@ def sign_in():  # pragma: no cover
     url = os.getenv("LOGIN_DOT_GOV_INITIAL_SIGNIN_URL")
 
     nonce = secrets.token_urlsafe()
-    session["nonce"] = nonce
+    redis_key = f"-{unquote(nonce)}"
+    redis_client.set(redis_key, nonce)
 
     # handle unit tests
     if url is not None:

app/notify_client/invite_api_client.py

@@ -1,3 +1,7 @@
+import secrets
+from urllib.parse import unquote
+
+from app import redis_client
 from app.notify_client import NotifyAdminAPIClient, _attach_current_user, cache
 from app.utils.user_permissions import (
     all_ui_permissions,
@@ -32,6 +36,13 @@ class InviteApiClient(NotifyAdminAPIClient):
             "folder_permissions": folder_permissions,
         }
         data = _attach_current_user(data)
+
+        # make and store the nonce
+        nonce = secrets.token_urlsafe()
+        redis_key = f"login-nonce-{unquote(nonce)}"
+        redis_client.set(f"{redis_key}", nonce)  # save the nonce to redis.
+        data["nonce"] = nonce  # This is passed to api for the invite url.
+
         resp = self.post(url=f"/service/{service_id}/invite", data=data)
         return resp["data"]
 

app/utils/login.py

@@ -1,6 +1,10 @@
+import json
+import os
 from functools import wraps
 
-from flask import redirect, request, session, url_for
+import jwt
+import requests
+from flask import current_app, redirect, request, session, url_for
 
 from app.models.user import User
 from app.utils.time import is_less_than_days_ago
@@ -57,3 +61,32 @@ def is_safe_redirect_url(target):
         redirect_url.scheme in ("http", "https")
         and host_url.netloc == redirect_url.netloc
     )
+
+
+def get_id_token(json_data):
+    """Decode and return the id_token."""
+    client_id = os.getenv("LOGIN_DOT_GOV_CLIENT_ID")
+    certs_url = os.getenv("LOGIN_DOT_GOV_CERTS_URL")
+
+    try:
+        encoded_id_token = json_data["id_token"]
+    except KeyError as e:
+        current_app.logger.exception(f"Error when getting id token {json_data}")
+        raise KeyError(f"'access_token' {request.json()}") from e
+
+    # Getting Login.gov signing keys for unpacking the id_token correctly.
+    jwks = requests.get(certs_url, timeout=5).json()
+    public_keys = {
+        jwk["kid"]: {
+            "key": jwt.algorithms.RSAAlgorithm.from_jwk(json.dumps(jwk)),
+            "algo": jwk["alg"],
+        }
+        for jwk in jwks["keys"]
+    }
+    kid = jwt.get_unverified_header(encoded_id_token)["kid"]
+    pub_key = public_keys[kid]["key"]
+    algo = public_keys[kid]["algo"]
+    id_token = jwt.decode(
+        encoded_id_token, pub_key, audience=client_id, algorithms=[algo]
+    )
+    return id_token

tests/app/notify_client/test_invite_client.py

@@ -25,11 +25,16 @@ def test_client_creates_invite(
                     "created_at",
                     "auth_type",
                     "folder_permissions",
+                    "nonce",
                 }
             )
         },
     )
 
+    mock_token_urlsafe = mocker.patch("secrets.token_urlsafe")
+    fake_nonce = "1234567890"
+    mock_token_urlsafe.return_value = fake_nonce
+
     invite_api_client.create_invite(
         "12345", "67890", "test@example.com", {"send_messages"}, "sms_auth", [fake_uuid]
     )
@@ -45,6 +50,7 @@ def test_client_creates_invite(
             "permissions": "send_emails,send_texts",
             "invite_link_host": "http://localhost:6012",
             "folder_permissions": [fake_uuid],
+            "nonce": fake_nonce,
         },
     )
 

tests/app/utils/test_login.py

@@ -20,3 +20,23 @@ def test_email_needs_revalidating(
 ):
     api_user_active["email_access_validated_at"] = email_access_validated_at
     assert email_needs_revalidating(User(api_user_active)) == expected_result
+
+
+def test_get_id_token(mocker):
+    mock_request_get = mocker.patch("requests.get")
+    fake_kid = "fake"
+    mock_request_get.return_value.json.return_value = {
+        "keys": [{"kid": fake_kid, "alg": "whatever"}]
+    }
+    mock_dumps = mocker.patch("json.dumps")
+    mock_dumps.return_value = "Not really JSON"
+    mock_from_jwk = mocker.patch("jwt.algorithms.RSAAlgorithm.from_jwk")
+    mock_from_jwk.return_value = "Fake fake fake"
+    mock_get_unverified_header = mocker.patch("jwt.get_unverified_header")
+    mock_get_unverified_header.return_value.__getitem__.return_value = fake_kid
+    mock_decode = mocker.patch("jwt.decode")
+    from app.utils import login
+
+    login.get_id_token({"id_token": "Not a token"})
+
+    assert mock_decode.called