refactor auth module, support for renew auth cookie if token expired #947

Author: vgrem

office365/runtime/auth/auth_cookies.py

@@ -0,0 +1,48 @@
+import json
+from typing import Dict, Optional
+
+
+class AuthCookies:
+    def __init__(self, cookies):
+        # type: (Dict) -> None
+        self._cookies = cookies
+
+    @property
+    def is_valid(self) -> bool:
+        """Validates authorization cookies."""
+        return bool(self._cookies) and (
+            self.fed_auth is not None or self.spo_idcrl is not None
+        )
+
+    @property
+    def cookie_header(self):
+        """Converts stored cookies into an HTTP Cookie header string."""
+        return "; ".join(f"{key}={val}" for key, val in self._cookies.items())
+
+    @property
+    def fed_auth(self):
+        # type: () -> Optional[str]
+        """Returns the Primary authentication token."""
+        return self._cookies.get("FedAuth", None)
+
+    @property
+    def spo_idcrl(self):
+        # type: () -> Optional[str]
+        """Returns the secondary authentication token. (SharePoint Online Identity CRL)"""
+        return self._cookies.get("SPOIDCRL", None)
+
+    @property
+    def rt_fa(self):
+        # type: () -> Optional[str]
+        """Returns the refresh token for Federated Authentication."""
+        return self._cookies.get("rtFa", None)
+
+    def to_json(self):
+        """Serializes cookies to JSON format."""
+        return json.dumps(self._cookies, indent=2)
+
+    @classmethod
+    def from_json(cls, json_data):
+        """Deserializes cookies from JSON format."""
+        cookies_dict = json.loads(json_data)
+        return cls(cookies_dict)

office365/runtime/auth/providers/acs_token_provider.py

@@ -26,13 +26,14 @@ def __init__(self, url, client_id, client_secret, environment="commercial"):
         self.SharePointPrincipal = "00000003-0000-0ff1-ce00-000000000000"
         self._client_id = client_id
         self._client_secret = client_secret
-        self._cached_token = None
+        self._cached_token = None  # type: Optional[TokenResponse]
         self._environment = environment
 
     def authenticate_request(self, request):
         # type: (RequestOptions) -> None
-        self._ensure_app_only_access_token()
-        request.set_header("Authorization", self._get_authorization_header())
+        if self._cached_token is None:
+            self._cached_token = self.get_app_only_access_token()
+        request.set_header("Authorization", self._cached_token.authorization_header)
 
     def get_app_only_access_token(self):
         """Retrieves an app-only access token from ACS"""
@@ -48,11 +49,6 @@ def get_app_only_access_token(self):
             )
             raise ValueError(self.error)
 
-    def _ensure_app_only_access_token(self):
-        if self._cached_token is None:
-            self._cached_token = self.get_app_only_access_token()
-        return self._cached_token and self._cached_token.is_valid
-
     def _get_app_only_access_token(self, target_host, target_realm):
         """
         Retrieves an app-only access token from ACS to call the specified principal
@@ -86,11 +82,6 @@ def _get_app_only_access_token(self, target_host, target_realm):
     def _get_realm_from_target_url(self):
         """Get the realm for the URL"""
         response = requests.head(url=self.url, headers={"Authorization": "Bearer"})
-        return self.process_realm_response(response)
-
-    @staticmethod
-    def process_realm_response(response):
-        # type: (requests.Response) -> Optional[str]
         header_key = "WWW-Authenticate"
         if header_key in response.headers:
             auth_values = response.headers[header_key].split(",")
@@ -116,6 +107,3 @@ def get_security_token_service_url(realm, environment):
                     realm
                 )
             )
-
-    def _get_authorization_header(self):
-        return "Bearer {0}".format(self._cached_token.accessToken)

office365/runtime/auth/providers/saml_token_provider.py

@@ -1,12 +1,15 @@
 import os
 import uuid
+from datetime import datetime, timezone
+from typing import Optional
 from xml.dom import minidom
 from xml.etree import ElementTree
 
 import requests
 import requests.utils
 
 import office365.logger
+from office365.runtime.auth.auth_cookies import AuthCookies
 from office365.runtime.auth.authentication_provider import AuthenticationProvider
 from office365.runtime.auth.sts_profile import STSProfile
 from office365.runtime.auth.user_realm_info import UserRealmInfo
@@ -20,23 +23,17 @@ def resolve_base_url(url):
     return parts[0] + "://" + host_name
 
 
-def xml_escape(s_val):
-    s_val = s_val.replace("&", "&")
-    s_val = s_val.replace("<", "&lt;")
-    s_val = s_val.replace(">", "&gt;")
-    s_val = s_val.replace('"', "&quot;")
-    s_val = s_val.replace("'", "&apos;")
-    return s_val
+def string_escape(value):
+    value = value.replace("&", "&amp;")
+    value = value.replace("<", "&lt;")
+    value = value.replace(">", "&gt;")
+    value = value.replace('"', "&quot;")
+    value = value.replace("'", "&apos;")
+    return value
 
 
-def is_valid_auth_cookies(values):
-    """
-    Validates authorization cookies
-    """
-    return any(values) and (
-        values.get("FedAuth", None) is not None
-        or values.get("SPOIDCRL", None) is not None
-    )
+def datetime_escape(value):
+    return value.isoformat("T")[:-9] + "Z"
 
 
 class SamlTokenProvider(AuthenticationProvider, office365.logger.LoggerContext):
@@ -60,7 +57,7 @@ def __init__(self, url, username, password, browser_mode, environment="commercia
         self.error = ""
         self._username = username
         self._password = password
-        self._cached_auth_cookies = None
+        self._cached_auth_cookies = None  # type: Optional[AuthCookies]
         self.__ns_prefixes = {
             "S": "{http://www.w3.org/2003/05/soap-envelope}",
             "s": "{http://www.w3.org/2003/05/soap-envelope}",
@@ -82,24 +79,20 @@ def authenticate_request(self, request):
         Authenticate request handler
         """
         logger = self.logger(self.authenticate_request.__name__)
-        self.ensure_authentication_cookie()
-        logger.debug_secrets(self._cached_auth_cookies)
-        cookie_header_value = "; ".join(
-            [
-                "=".join([key, str(val)])
-                for key, val in self._cached_auth_cookies.items()
-            ]
-        )
-        request.set_header("Cookie", cookie_header_value)
 
-    def ensure_authentication_cookie(self):
-        if self._cached_auth_cookies is None:
+        request_time = datetime.now(timezone.utc)
+        if (
+            self._cached_auth_cookies is None
+            or request_time >= self._sts_profile.expires
+        ):
+            self._sts_profile.reset()
             self._cached_auth_cookies = self.get_authentication_cookie()
-        return True
+        logger.debug_secrets(self._cached_auth_cookies)
+        request.set_header("Cookie", self._cached_auth_cookies.cookie_header)
 
     def get_authentication_cookie(self):
         """Acquire authentication cookie"""
-        logger = self.logger(self.ensure_authentication_cookie.__name__)
+        logger = self.logger(self.get_authentication_cookie.__name__)
         logger.debug("get_authentication_cookie called")
 
         try:
@@ -140,10 +133,10 @@ def _acquire_service_token_from_adfs(self, adfs_url):
             {
                 "auth_url": adfs_url,
                 "message_id": str(uuid.uuid4()),
-                "username": xml_escape(self._username),
-                "password": xml_escape(self._password),
-                "created": self._sts_profile.created,
-                "expires": self._sts_profile.expires,
+                "username": string_escape(self._username),
+                "password": string_escape(self._password),
+                "created": datetime_escape(self._sts_profile.created),
+                "expires": datetime_escape(self._sts_profile.expires),
                 "issuer": self._sts_profile.tokenIssuer,
             },
         )
@@ -192,11 +185,11 @@ def _acquire_service_token(self):
             "SAML.xml",
             {
                 "auth_url": self._sts_profile.authorityUrl,
-                "username": xml_escape(self._username),
-                "password": xml_escape(self._password),
+                "username": string_escape(self._username),
+                "password": string_escape(self._password),
                 "message_id": str(uuid.uuid4()),
-                "created": self._sts_profile.created,
-                "expires": self._sts_profile.expires,
+                "created": datetime_escape(self._sts_profile.created),
+                "expires": datetime_escape(self._sts_profile.expires),
                 "issuer": self._sts_profile.tokenIssuer,
             },
         )
@@ -297,9 +290,9 @@ def _get_authentication_cookie(self, security_token, federated=False):
                 },
             )
         logger.debug_secrets("session.cookies: %s", session.cookies)
-        cookies = requests.utils.dict_from_cookiejar(session.cookies)
+        cookies = AuthCookies(requests.utils.dict_from_cookiejar(session.cookies))
         logger.debug_secrets("cookies: %s", cookies)
-        if not is_valid_auth_cookies(cookies):
+        if not cookies.is_valid:
             self.error = (
                 "An error occurred while retrieving auth cookies from {0}".format(
                     self._sts_profile.signin_page_url

office365/runtime/auth/sts_profile.py

@@ -5,9 +5,7 @@
 
 class STSProfile(object):
     def __init__(self, authority_url, environment):
-        """
-        :type authority_url: str
-        """
+        # type: (str, str) -> None
         self.authorityUrl = authority_url
         if environment == "GCCH":
             self.serviceUrl = "https://login.microsoftonline.us"
@@ -16,13 +14,15 @@ def __init__(self, authority_url, environment):
         self.securityTokenServicePath = "extSTS.srf"
         self.userRealmServicePath = "GetUserRealm.srf"
         self.tokenIssuer = "urn:federation:MicrosoftOnline"
-        now = datetime.now(tz=timezone.utc)
-        self.created = now.astimezone(timezone.utc).isoformat("T")[:-9] + "Z"
-        self.expires = (now + timedelta(minutes=10)).astimezone(timezone.utc).isoformat(
-            "T"
-        )[:-9] + "Z"
+        self.created = datetime.now(tz=timezone.utc)
+        self.expires = self.created + timedelta(minutes=30)
         self.signInPage = "_forms/default.aspx?wa=wsignin1.0"
 
+    def reset(self):
+        """Renew the expiration time."""
+        self.created = datetime.now(tz=timezone.utc)
+        self.expires = self.created + timedelta(minutes=30)
+
     @property
     def tenant(self):
         return urlparse(self.authorityUrl).netloc

office365/runtime/auth/token_response.py

@@ -9,6 +9,10 @@ def __init__(self, access_token=None, token_type=None, **kwargs):
     def is_valid(self):
         return self.accessToken is not None and self.tokenType == "Bearer"
 
+    @property
+    def authorization_header(self):
+        return "Bearer {0}".format(self.accessToken)
+
     @staticmethod
     def from_json(value):
         error = value.get("error", None)