roadtx: selenium improvements and msa ticket support for device regis…

…tration

roadlib/roadtools/roadlib/deviceauth.py

@@ -13,6 +13,8 @@
 import time
 import warnings
 import datetime
+import uuid
+import urllib3
 from cryptography.hazmat.primitives import serialization, padding, hashes
 from cryptography.hazmat.primitives.asymmetric import rsa
 from cryptography.hazmat.primitives.asymmetric import padding as apadding
@@ -23,8 +25,8 @@
 from cryptography.x509.oid import NameOID
 from cryptography.utils import CryptographyDeprecationWarning
 from roadtools.roadlib.auth import Authentication, get_data, AuthenticationException
-
 warnings.filterwarnings("ignore", category=CryptographyDeprecationWarning)
+warnings.filterwarnings("ignore", category=urllib3.exceptions.InsecureRequestWarning)
 
 
 class DeviceAuthentication():
@@ -164,7 +166,37 @@ def create_pubkey_blob_from_key(self, key):
         pubkeycngblob = b''.join(header)+exponent_as_bytes+modulus_as_bytes
         return pubkeycngblob
 
-    def register_device(self, access_token, jointype=0, certout=None, privout=None, device_type=None, device_name=None, os_version=None):
+    def create_public_jwk_from_key(self, key, for_registration=False):
+        """
+        Convert a key (or certificate) to JWK public numbers
+        https://www.rfc-editor.org/rfc/rfc7517
+        """
+        pubkey = key.public_key()
+        pubnumbers = pubkey.public_numbers()
+
+        # From python docs https://docs.python.org/3/library/stdtypes.html#int.to_bytes
+        exponent_as_bytes = pubnumbers.e.to_bytes((pubnumbers.e.bit_length() + 7) // 8, byteorder='big')
+        modulus_as_bytes = pubnumbers.n.to_bytes((pubnumbers.n.bit_length() + 7) // 8, byteorder='big')
+
+
+        if for_registration:
+            # Registration expects additional parameters and different encoding (regular base64 instead of urlsafe)
+            jwk = {
+                'kty': 'RSA',
+                'e': base64.b64encode(exponent_as_bytes).decode('utf-8'),
+                'n': base64.b64encode(modulus_as_bytes).decode('utf-8'),
+                'alg': 'RS256',
+                'kid': str(uuid.uuid4()).upper()
+            }
+        else:
+            jwk = {
+                'kty': 'RSA',
+                'e': base64.urlsafe_b64encode(exponent_as_bytes).decode('utf-8'),
+                'n': base64.urlsafe_b64encode(modulus_as_bytes).decode('utf-8'),
+            }
+        return json.dumps(jwk, separators=(',', ':'))
+
+    def register_device(self, access_token, jointype=0, certout=None, privout=None, device_type=None, device_name=None, os_version=None, deviceticket=None):
         """
         Registers a device in Azure AD. Requires an access token to the device registration service.
         """
@@ -209,24 +241,45 @@ def register_device(self, access_token, jointype=0, certout=None, privout=None,
 
         pubkeycngblob = base64.b64encode(self.create_pubkey_blob_from_key(key))
 
-        data = {
-            "CertificateRequest":
-                {
-                    "Type": "pkcs10",
-                    "Data": certbytes.decode('utf-8')
-                },
-            "TransportKey": pubkeycngblob.decode('utf-8'),
-            # Can likely be edited to anything, are not validated afaik
-            "TargetDomain": "iminyour.cloud",
-            "DeviceType": device_type,
-            "OSVersion": os_version,
-            "DeviceDisplayName": device_name,
-            "JoinType": jointype,
-            "attributes": {
-                "ReuseDevice": "true",
-                "ReturnClientSid": "true"
+
+        if device_type.lower() == 'macos':
+            data = {
+              "DeviceDisplayName" : device_name,
+              "CertificateRequest" : {
+                "Type" : "pkcs10",
+                "Data" : certbytes.decode('utf-8')
+              },
+              "OSVersion" : "12.2.0",
+              "TargetDomain" : "iminyour.cloud",
+              "AikCertificate" : "",
+              "DeviceType" : "MacOS",
+              "TransportKey" : base64.b64encode(self.create_public_jwk_from_key(key, True).encode('utf-8')).decode('utf-8'),
+              "JoinType" : jointype,
+              "AttestationData" : ""
             }
-        }
+        else:
+            data = {
+                "CertificateRequest":
+                    {
+                        "Type": "pkcs10",
+                        "Data": certbytes.decode('utf-8')
+                    },
+                "TransportKey": pubkeycngblob.decode('utf-8'),
+                # Can likely be edited to anything, are not validated afaik
+                "TargetDomain": "iminyour.cloud",
+                "DeviceType": device_type,
+                "OSVersion": os_version,
+                "DeviceDisplayName": device_name,
+                "JoinType": jointype,
+                "attributes": {
+                    "ReuseDevice": "true",
+                    "ReturnClientSid": "true"
+                }
+            }
+            # Add device ticket if requested
+            if deviceticket:
+                data['attributes']['MSA-DDID'] = base64.b64encode(deviceticket.encode('utf-8')).decode('utf-8')
+
 
         headers = {
             'Authorization': f'Bearer {access_token}',

roadlib/setup.py

@@ -1,6 +1,6 @@
 from setuptools import setup, find_namespace_packages
 setup(name='roadlib',
-      version='0.15.2',
+      version='0.16.0',
       description='ROADtools common components library',
       author='Dirk-jan Mollema',
       author_email='[email protected]',

roadtx/roadtools/roadtx/main.py

@@ -38,6 +38,7 @@ def main():
     device_parser.add_argument('--access-token', action='store', help='Access token for device registration service. If not specified, taken from .roadtools_auth')
     device_parser.add_argument('--device-type', action='store', help='Device OS type (default: Windows)')
     device_parser.add_argument('--os-version', action='store', help='Device OS version (default: 10.0.19041.928)')
+    device_parser.add_argument('--deviceticket', action='store', help='Device MSA ticket to match with existing device')
 
     # Construct hybrid device module
     hdevice_parser = subparsers.add_parser('hybriddevice', help='Join an on-prem device to Azure AD')
@@ -319,6 +320,9 @@ def main():
     browserprtauth_parser.add_argument('-k', '--keep-open',
                                        action='store_true',
                                        help='Do not close the browser window after timeout. Useful if you want to browse online apps with the obtained credentials')
+    browserprtauth_parser.add_argument('--capture-code',
+                                       action='store_true',
+                                       help='Do not attempt to redeem any authentication code but print it instead')
 
     # Interactive auth using Selenium - inject PRT to other user
     injauth_parser = subparsers.add_parser('browserprtinject', help='Selenium based auth with automatic PRT injection. Can be used with other users to add device state to session')
@@ -431,7 +435,7 @@ def main():
                 jointype = 0
             else:
                 jointype = 4
-            deviceauth.register_device(tokenobject['accessToken'], jointype=jointype, certout=args.cert_pem, privout=args.key_pem, device_type=args.device_type, device_name=args.name, os_version=args.os_version)
+            deviceauth.register_device(tokenobject['accessToken'], jointype=jointype, certout=args.cert_pem, privout=args.key_pem, device_type=args.device_type, device_name=args.name, os_version=args.os_version, deviceticket=args.deviceticket)
         elif args.action == 'delete':
             if not deviceauth.loadcert(args.cert_pem, args.key_pem):
                 return
@@ -591,7 +595,9 @@ def main():
         if not service:
             return
         selauth.driver = selauth.get_webdriver(service, intercept=True)
-        if not selauth.selenium_login_with_prt(url, keep=args.keep_open, prtcookie=args.prt_cookie):
+        if not selauth.selenium_login_with_prt(url, keep=args.keep_open, prtcookie=args.prt_cookie, capture=args.capture_code):
+            return
+        if args.capture_code:
             return
         auth.outfile = args.tokenfile
         auth.save_tokens(args)

roadtx/roadtools/roadtx/selenium.py

@@ -143,7 +143,7 @@ def selenium_login(self, url, identity=None, password=None, otpseed=None, keep=F
                 raise AuthenticationException('Authentication did not complete within time limit')
         return False
 
-    def selenium_login_with_prt(self, url, identity=None, password=None, otpseed=None, keep=False, prtcookie=None):
+    def selenium_login_with_prt(self, url, identity=None, password=None, otpseed=None, keep=False, prtcookie=None, capture=False):
         '''
         Selenium login with PRT injection.
         '''
@@ -172,7 +172,7 @@ def interceptor(request):
                                                                            self.deviceauth.session_key)
                         request.headers['X-Ms-Refreshtokencredential'] = cookie
         self.driver.request_interceptor = interceptor
-        return self.selenium_login(url, identity, password, otpseed, keep=keep)
+        return self.selenium_login(url, identity, password, otpseed, keep=keep, capture=capture)
 
     def selenium_enrich_prt(self, url, otpseed=None):
         '''

roadtx/setup.py

@@ -1,6 +1,6 @@
 from setuptools import setup
 setup(name='roadtx',
-      version='1.2.0',
+      version='1.2.1',
       description='ROADtools Token eXchange',
       author='Dirk-jan Mollema',
       author_email='[email protected]',
@@ -17,7 +17,7 @@
       ],
       packages=['roadtools.roadtx'],
       install_requires=[
-          'roadlib>=0.15',
+          'roadlib>=0.16',
           'requests',
           'selenium',
           'selenium-wire',