Merge pull request #124 from GSA-TTS/issue-113-aanand-login-gov-auth

Changes for login.gov authentication

nad_ch/application/interfaces.py

@@ -1,4 +1,4 @@
-from typing import Optional, Protocol, Dict
+from typing import Optional, Protocol, Dict, Union
 from nad_ch.application.dtos import DownloadResult
 from nad_ch.core.repositories import (
     DataProducerRepository,
@@ -37,20 +37,26 @@ def run_load_and_validate(
 
 
 class Authentication(Protocol):
-    def fetch_oauth2_token(self, provider_name: str, code: str) -> str | None: ...
+    def fetch_oauth2_token(self, provider_name: str, code: str) -> Optional[str]: ...
 
     def fetch_user_email_from_login_provider(
         self, provider_name: str, oauth2_token: str
-    ) -> str | list[str] | None: ...
+    ) -> Union[str, list[str], None]: ...
 
     def get_logout_url(self, provider_name: str) -> str: ...
 
-    def make_login_url(self, provider_name: str, state_token: str) -> str | None: ...
+    def make_login_url(
+        self,
+        provider_name: str,
+        state_token: str,
+        acr_values: Optional[str] = None,
+        nonce: Optional[str] = None
+    ) -> Optional[str]: ...
 
-    def make_logout_url(self, provider_name: str) -> str | None: ...
+    def make_logout_url(self, provider_name: str) -> Optional[str]: ...
 
     def user_email_address_has_permitted_domain(
-        self, email: str | list[str]
+        self, email: Union[str, list[str]]
     ) -> bool: ...
 
 

nad_ch/application/use_cases/auth.py

@@ -33,9 +33,15 @@ def get_or_create_user(ctx: ApplicationContext, provider_name: str, email: str)
 
 
 def get_logged_in_user_redirect_url(
-    ctx: ApplicationContext, provider_name: str, state_token: str
+    ctx: ApplicationContext,
+    provider_name: str,
+    state_token: str,
+    acr_values: str = None,
+    nonce: str = None,
 ) -> str | None:
-    return ctx.auth.make_login_url(provider_name, state_token)
+    return ctx.auth.make_login_url(
+        provider_name, state_token, acr_values=acr_values, nonce=nonce
+    )
 
 
 def get_logged_out_user_redirect_url(

nad_ch/config/base.py

@@ -1,30 +1,31 @@
 import os
+import secrets
 from dotenv import load_dotenv
 
-
 load_dotenv()
 
-
 APP_ENV = os.getenv("APP_ENV")
 PORT = os.getenv("PORT", 5000)
 ALLOWED_LOGIN_DOMAINS = ["dot.gov", "gsa.gov"]
 if os.environ.get("LOCAL_ALLOWED_DOMAIN"):
     ALLOWED_LOGIN_DOMAINS.append(os.environ.get("LOCAL_ALLOWED_DOMAIN"))
 CALLBACK_URL_SCHEME = os.getenv("CALLBACK_URL_SCHEME", "http")
 OAUTH2_CONFIG = {
-    "cloudgov": {
-        "client_id": os.environ.get("CLOUDGOV_CLIENT_ID"),
-        "client_secret": os.environ.get("CLOUDGOV_CLIENT_SECRET"),
-        "authorize_url": "https://login.fr.cloud.gov/oauth/authorize",
-        "token_url": "https://uaa.fr.cloud.gov/oauth/token",
-        "logout_url": "https://login.fr.cloud.gov/logout.do",
+    "logingov": {
+        "client_id": os.getenv("LOGINGOV_CLIENT_ID"),
+        "authorize_url": "https://idp.int.identitysandbox.gov/openid_connect/authorize",
+        "token_url": "https://idp.int.identitysandbox.gov/api/openid_connect/token",
+        "logout_url": "https://idp.int.identitysandbox.gov/openid_connect/logout",
         "userinfo": {
-            "url": "access_token",
+            "url": "https://idp.int.identitysandbox.gov/api/openid_connect/userinfo",
             "email": lambda json: json["email"],
         },
-        "scopes": ["openid"],
-    },
-    "logingov": {
-        # login.gov configuration details go here
-    },
+        "private_key_jwt": {
+            "key": os.getenv("LOGINGOV_PRIVATE_KEY"),
+            "alg": "RS256"
+        },
+        "acr_values": "http://idmanagement.gov/ns/assurance/ial/1",
+        "scopes": ["openid", "email", "profile"],
+        "nonce": lambda: secrets.token_urlsafe(64),
+    }
 }

nad_ch/controllers/web/routes/auth.py

@@ -74,13 +74,21 @@ def oauth2_authorize(provider: str):
         return redirect(url_for(login_view))
 
     state_token = secrets.token_urlsafe(16)
+    nonce = secrets.token_urlsafe(32)
+
+    redirect_url = get_logged_in_user_redirect_url(
+        g.ctx,
+        provider,
+        state_token,
+        nonce=nonce,
+        acr_values="http://idmanagement.gov/ns/assurance/ial/1",
+    )
 
-    redirect_url = get_logged_in_user_redirect_url(g.ctx, provider, state_token)
     if not redirect_url:
         abort(404)
 
     session["oauth2_state"] = state_token
-
+    session["oauth2_nonce"] = nonce
     return redirect(redirect_url)
 
 

nad_ch/controllers/web/templates/_layouts/sidebar.html

@@ -43,7 +43,7 @@
       </li>
       {% if current_user.is_authenticated %}
       <li class="usa-sidenav__item">
-        <a href="{{ url_for('auth.logout_provider', provider='cloudgov') }}"
+        <a href="{{ url_for('auth.logout_provider', provider='logingov') }}"
           >Logout</a
         >
       </li>

nad_ch/controllers/web/templates/index.html

@@ -7,9 +7,9 @@
 <div>
   <p>Log in to begin.</p>
   <a
-    href="{{ url_for('auth.oauth2_authorize', provider='cloudgov') }}"
+    href="{{ url_for('auth.oauth2_authorize', provider='logingov') }}"
     class="usa-button"
-    >Login with cloud.gov</a
+    >Sign in with login.gov</a
   >
 </div>
 {% endif %} {% endblock %}

nad_ch/infrastructure/auth.py

@@ -1,8 +1,14 @@
+import logging
+import time
+import uuid
+from datetime import timedelta
+
 from flask import url_for
 import jwt
 import requests
 from urllib.parse import urlencode
 from nad_ch.application.interfaces import Authentication
+from authlib.jose import jwt as jose_jwt
 
 
 class AuthenticationImplementation(Authentication):
@@ -18,22 +24,33 @@ def fetch_oauth2_token(self, provider_name: str, code: str) -> str | None:
         if not provider_config:
             return None
 
-        token_url = provider_config["token_url"]
+        private_key = provider_config["private_key_jwt"]["key"]
+        private_key = private_key.replace(r"\n", "\n")
+
+        payload = {
+            "iss": provider_config["client_id"],
+            "sub": provider_config["client_id"],
+            "aud": provider_config["token_url"],
+            "jti": str(uuid.uuid4()),
+            "exp": int(time.time()) + timedelta(minutes=30).seconds,
+        }
+
+        header = {"alg": provider_config["private_key_jwt"]["alg"]}
+
+        jws = jose_jwt.encode(header=header, payload=payload, key=private_key)
+        signed_jwt = jws.decode("utf-8")
+
         request_data = {
-            "client_id": provider_config["client_id"],
-            "client_secret": provider_config["client_secret"],
             "code": code,
             "grant_type": "authorization_code",
-            "redirect_uri": url_for(
-                "auth.oauth2_callback",
-                provider=provider_name,
-                _scheme=self._callback_url_scheme,
-                _external=True,
+            "client_assertion_type": (
+                "urn:ietf:params:oauth:client-assertion-type:jwt-bearer"
             ),
+            "client_assertion": signed_jwt,
         }
 
         response = requests.post(
-            token_url,
+            provider_config["token_url"],
             data=request_data,
             headers={"Accept": "application/json"},
             timeout=4,
@@ -63,9 +80,23 @@ def fetch_user_email_from_login_provider(
             )
 
             return decoded_data.get("email")
-        else:
-            # TODO handle login.gov case here
-            pass
+
+        user_info_url = provider_config["userinfo"].get("url")
+        if user_info_url:
+            headers = {
+                "Authorization": f"Bearer {oauth2_token}",
+                "Accept": "application/json",
+            }
+            try:
+                response = requests.get(user_info_url, headers=headers)
+                response.raise_for_status()
+                user_info = response.json()
+
+                email = user_info.get("email")
+                return email
+            except requests.RequestException as e:
+                logging.error(f"Failed to fetch user info: {e}")
+                return None
 
         return None
 
@@ -76,25 +107,36 @@ def get_logout_url(self, provider_name: str) -> str | None:
 
         return provider_config["logout_url"]
 
-    def make_login_url(self, provider_name: str, state_token: str) -> str | None:
+    def make_login_url(
+        self,
+        provider_name: str,
+        state_token: str,
+        acr_values: str = None,
+        nonce: str = None,
+    ) -> str | None:
         provider_config = self._providers[provider_name]
         if not provider_config:
             return None
 
-        query_string = urlencode(
-            {
-                "client_id": provider_config["client_id"],
-                "redirect_uri": url_for(
-                    "auth.oauth2_callback",
-                    provider=provider_name,
-                    _scheme=self._callback_url_scheme,
-                    _external=True,
-                ),
-                "response_type": "code",
-                "scope": " ".join(provider_config["scopes"]),
-                "state": state_token,
-            },
-        )
+        query_params = {
+            "client_id": provider_config["client_id"],
+            "redirect_uri": url_for(
+                "auth.oauth2_callback",
+                provider=provider_name,
+                _scheme=self._callback_url_scheme,
+                _external=True,
+            ),
+            "response_type": "code",
+            "scope": " ".join(provider_config["scopes"]),
+            "state": state_token,
+        }
+
+        if acr_values:
+            query_params["acr_values"] = acr_values
+        if nonce:
+            query_params["nonce"] = nonce
+
+        query_string = urlencode(query_params)
 
         return provider_config["authorize_url"] + "?" + query_string