Microsoft auth update

sso_email_check.py

@@ -3,6 +3,10 @@
 from sso_utils import env_var, to_list
 from email_helper import email_parts
 
+ENVIRONMENT = env_var("ENVIRONMENT", "development")
+IS_PROD = ENVIRONMENT.lower().startswith("prod")
+DEBUG = not IS_PROD
+
 
 def valid_email(email_input, client: dict = {}, debug: bool = False) -> dict:
     res = {"valid": False, "auth_type": None, "user_type": None}
@@ -119,6 +123,10 @@ def get_auth_type(email) -> str:
     if email == "[email protected]":
         return "email"
 
+    if not IS_PROD:
+        if email.endswith("@ncsc.gov.uk"):
+            return "microsoft"
+
     if email.endswith("@digital.cabinet-office.gov.uk"):
         return "google"
 

sso_microsoft_auth.py

@@ -7,6 +7,7 @@
 import traceback
 import requests
 import hashlib
+import ssl
 
 from jwt import PyJWKClient
 
@@ -16,6 +17,7 @@ def random_sha256() -> str:
 
 
 class MicrosoftAuth:
+    errored = False
     dev_mode = False
     _client_id: str = None
     _client_secret: str = None
@@ -38,7 +40,7 @@ def __init__(
         self,
         client_id: str = None,
         client_secret: str = None,
-        discovery_document_url: str = "https://login.microsoftonline.com/common/.well-known/openid-configuration",
+        discovery_document_url: str = "https://login.microsoftonline.com/common/v2.0/.well-known/openid-configuration",
         dev_mode: bool = False,
     ):
         self.setup(client_id, client_secret, discovery_document_url)
@@ -54,7 +56,7 @@ def reset(self, dev_mode: bool = False):
         self.token_endpoint = None
         self.jwks_uri = None
 
-        self.scopes = ["openid", "User.ReadBasic.All"]
+        self.scopes = ["openid", "email", "profile"]
 
         self.dev_mode = dev_mode
 
@@ -63,16 +65,15 @@ def reset(self, dev_mode: bool = False):
     def get_oidc_config(self) -> dict:
         if not self._microsoft_fetch_is_error and not self._microsoft_oidc_config:
             try:
-                with urllib.request.urlopen(
-                    self.discovery_document_url, timeout=3
-                ) as url:
-                    if url:
-                        data = json.load(url)
-                        if data and "issuer" in data:
-                            self._microsoft_oidc_config = data
+                resp = requests.get(self.discovery_document_url, timeout=3)
+                if resp.status_code == 200 and resp.json():
+                    data = resp.json()
+                    if data and "issuer" in data:
+                        self._microsoft_oidc_config = data
             except Exception as e:
                 self._microsoft_fetch_error_msg = str(e) + traceback.format_exc()
                 self._microsoft_fetch_is_error = True
+                self.errored = True
 
         return self._microsoft_oidc_config
 
@@ -93,6 +94,13 @@ def _init_config(self, oev=True):
 
     def is_ready(self) -> bool:
         starts = f"http{'s://' if not self.dev_mode else ''}"
+        print("is_ready:starts:", starts)
+        print(
+            "is_ready:endpoints:",
+            self.auth_endpoint,
+            self.token_endpoint,
+            self.jwks_uri,
+        )
         return 3 == [
             starts
             for x in [self.auth_endpoint, self.token_endpoint, self.jwks_uri]
@@ -169,7 +177,6 @@ def step_two_get_id_token_from_microsoft_url(
         state_to_compare: str = None,
         redirect_uri: str = None,
     ) -> dict:
-
         if not url:
             return {"error": True, "error_message": "Argument 'url' not set or empty"}
 
@@ -293,14 +300,22 @@ def verify_and_decode_id_token(self, id_token: str = None) -> tuple:
         data = {}
 
         try:
-            jwks_client = PyJWKClient(self.jwks_uri)
-            signing_key = jwks_client.get_signing_key_from_jwt(id_token)
+            # ssl_context = ssl.create_default_context()
+            # ssl_context.check_hostname = False
+            # ssl_context.verify_mode = ssl.CERT_NONE
+
+            # jwks_client = PyJWKClient(self.jwks_uri, ssl_context=ssl_context)
+            # signing_key = jwks_client.get_signing_key_from_jwt(id_token)
+
             data = jwt.decode(
                 id_token,
-                signing_key.key,
-                algorithms=["RS256"],
+                # apparently MS doesn't sign their JWTs...
+                verify=False,
+                # key=signing_key.key,
+                # algorithms=["RS256"],
+                algorithms=["none"],
                 audience=self._client_id,
-                options={"verify_exp": True},
+                options={"verify_exp": True, "verify_signature": False},
             )
         except Exception as e:
             return (True, str(e) + traceback.format_exc())

wsgi.py

@@ -82,8 +82,14 @@
 MICROSOFT_CLIENT_SECRET = env_var("MICROSOFT_CLIENT_SECRET")
 ma = None
 try:
-    ma = MicrosoftAuth(MICROSOFT_CLIENT_ID, MICROSOFT_CLIENT_SECRET)
-    jprint({"MicrosoftAuth": {"in_use": True}})
+    ma = MicrosoftAuth(MICROSOFT_CLIENT_ID, MICROSOFT_CLIENT_SECRET, dev_mode=DEBUG)
+    if not ma.errored:
+        jprint({"MicrosoftAuth": {"in_use": True}})
+    else:
+        jprint(
+            {"MicrosoftAuth": {"in_use": False, "error": ma._microsoft_fetch_error_msg}}
+        )
+        ma = None
 except Exception as e:
     jprint({"MicrosoftAuth": {"error": e, "in_use": False}})
 
@@ -631,7 +637,11 @@ def microsoft_callback():
             "error" in mr
             and mr["error"]
             and "error_message" in mr
-            and "login_required" in mr["error_message"]
+            and (
+                "login_required" in str(mr["error_message"])
+                or "not consented" in str(mr["error_message"])
+                or "interaction_required" in str(mr["error_message"])
+            )
         ):
             if "microsoft_retry" not in session or not session["microsoft_retry"]:
                 session.pop("microsoft_retry", None)
@@ -657,16 +667,16 @@ def microsoft_callback():
                 force_email=True,
             )
 
-        id_token = mr["id_token"]
+        id_token = mr.get("id_token", None)
 
-        if "amr" not in id_token or "mfa" not in id_token["amr"]:
+        if not id_token or not id_token.get("email", None):
             return return_sign_in(
                 is_error=True,
-                fail_message="Microsoft account missing multifactor authentication, please continue to try again with an email code",
+                fail_message="Microsoft account sign in failed, please continue to try again with an email code",
                 force_email=True,
             )
 
-        email = email_parts(id_token["upn"])
+        email = email_parts(id_token["email"])
         ve = valid_email(email, debug=DEBUG)
         if not ve["valid"]:
             return return_sign_in(
@@ -1564,6 +1574,7 @@ def signin():
         if ve["user_type"] is None or ve["user_type"] != "user":
             return returnError(403)
 
+        print("auth_type:", auth_type)
         session["email"] = email
 
         remember_me = False
@@ -1596,6 +1607,7 @@ def signin():
                 login_hint=email["email"],
                 domain_hint=email["domain"],
             )
+            print("MICROSOFT:", mr)
             if "error" in mr and mr["error"] == False and "url" in mr:
                 session["microsoft_state"] = mr["state"]
                 session["microsoft_nonce"] = mr["nonce"]