Certificates and joining enforce-secure-profile servers

quarry/net/auth.py

@@ -6,6 +6,7 @@
 from cryptography.hazmat.primitives.asymmetric.rsa import RSAPublicKey
 from twisted.internet import defer
 from quarry.net import http
+from quarry.types.certificate import CertificatePair
 from quarry.types.uuid import UUID
 
 if sys.version_info[0] == 2:
@@ -39,11 +40,13 @@ class Profile(object):
     online = True
     timeout = 30
 
-    def __init__(self, client_token, access_token, display_name, uuid):
+    def __init__(self, client_token, access_token, display_name, uuid, certificates=None):
         self.client_token = client_token
         self.access_token = access_token
         self.display_name = display_name
         self.uuid = uuid
+        self.certificates = CertificatePair.from_dict(certificates) if certificates else None
+        self.enable_signing = self.certificates is not None
 
     def join(self, digest, refresh=True):
         d1 = http.request(
@@ -75,6 +78,35 @@ def _callback(data):
         d1.addCallbacks(_callback, d0.errback)
         return d0
 
+    def _get_certificates(self):
+        d0 = defer.Deferred()
+
+        def _callback(data):
+            keyPair = data["keyPair"]
+            d0.callback(CertificatePair(keyPair["privateKey"], keyPair["publicKey"], data["publicKeySignature"],
+                                        data["publicKeySignatureV2"], data["expiresAt"]))
+
+        # TODO "post" is to force http.request to send a POST request, which shouldn't be required
+        d1 = self._request_services(b"player/certificates", post=True)
+        d1.addCallbacks(_callback, d0.errback)
+        return d0
+
+    def use_signing(self):
+        d0 = defer.Deferred()
+
+        def _callback(data):
+            self.certificates = data
+            self.enable_signing = True
+            d0.callback(self.certificates)
+
+        if not self.certificates or self.certificates.is_expired():
+            d1 = self._get_certificates()
+            d1.addCallbacks(_callback, d0.errback)
+            return d0
+        else:
+            self.enable_signing = True
+            return self.certificates
+
     def refresh(self):
         d0 = defer.Deferred()
 
@@ -100,7 +132,11 @@ def to_file(self, profiles_path=None):
                     self.uuid.to_hex(False): {
                         "displayName": self.display_name,
                         "accessToken": self.access_token,
-                        "uuid": self.uuid.to_hex(True)}}}, fd)
+                        "uuid": self.uuid.to_hex(True)
+                    }
+                },
+                "certificates": self.certificates.to_dict() if self.certificates else None
+            }, fd)
 
     @classmethod
     def from_credentials(cls, email, password):
@@ -130,9 +166,9 @@ def _errback(err):
         return d0
 
     @classmethod
-    def from_token(cls, client_token, access_token, display_name, uuid):
+    def from_token(cls, client_token, access_token, display_name, uuid, certificates=None):
         obj = cls(client_token, access_token,
-                  display_name, UUID.from_hex(uuid))
+                  display_name, UUID.from_hex(uuid), certificates=certificates)
         return obj.validate()
 
     @classmethod
@@ -144,6 +180,7 @@ def from_file(cls, display_name=None, uuid=None, profiles_path=None):
             data = json.load(fd)
 
         client_token = data["clientToken"]
+        certificates = data["certificates"]
 
         if display_name is None and uuid is None:
             uuid = data["selectedUser"]["profile"]
@@ -158,7 +195,7 @@ def from_file(cls, display_name=None, uuid=None, profiles_path=None):
                 if display_name and display_name != p_display_name:
                     continue
                 return cls.from_token(client_token, access_token,
-                                      p_display_name, p_uuid)
+                                      p_display_name, p_uuid, certificates=certificates)
 
     @classmethod
     def _from_response(cls, response):
@@ -176,6 +213,15 @@ def _request(cls, endpoint, **data):
             err_type=ProfileException,
             data=data)
 
+    def _request_services(self, endpoint, **data):
+        return http.request(
+            url=b"https://api.minecraftservices.com/" + endpoint,
+            timeout=self.timeout,
+            err_type=ProfileException,
+            data=data,
+            headers={"Authorization": [f"Bearer {self.access_token}"]}
+        )
+
     @classmethod
     def _get_profiles_path(cls):
         dot_minecraft = ".minecraft"

quarry/net/client.py

@@ -1,9 +1,17 @@
+import re
+
+import base64
+import os
+from OpenSSL import crypto as ssl_crypto
+
 from twisted.internet import reactor, protocol, defer
 from twisted.python import failure
+from quarry.types.certificate import CertificatePair
 
 from quarry.types.chat import Message
 from quarry.net.protocol import Factory, Protocol, ProtocolError, \
     protocol_modes_inv
+from quarry.net.auth import Profile, OfflineProfile
 from quarry.net import auth, crypto
 
 
@@ -41,15 +49,37 @@ def switch_protocol_mode(self, mode):
         elif mode == "login":
             # Send login start
             # TODO: Implement signatures/1.19.1 UUID sending
+            profile: Profile = self.factory.profile
+            can_use_signing_data = (
+                    profile.online
+                    and isinstance(profile, Profile)
+                    and profile.enable_signing
+                    and profile.certificates is not None
+            )
+            signature = [self.buff_type.pack("?", can_use_signing_data)]
+            uuid = [self.buff_type.pack("?", can_use_signing_data)]
+            if can_use_signing_data:
+                cert = profile.certificates
+                public_key = CertificatePair.convert_public_key(cert.public)
+                pk_len = len(public_key)
+                signature += [
+                    self.buff_type.pack("q", int(cert.expires.timestamp() * 1000)),  # Time the certificate expires
+                    self.buff_type.pack_varint(pk_len),
+                    self.buff_type.pack(f'{pk_len}s', public_key),  # Public Key (incl. Length)
+                    self.buff_type.pack_byte_array(base64.b64decode(cert.signature2)),  # Signature (incl. Length)
+                ]
+                uuid += [
+                    self.buff_type.pack_uuid(profile.uuid),
+                ]
             if self.protocol_version >= 760:  # 1.19.1+
                 self.send_packet("login_start",
                                  self.buff_type.pack_string(self.factory.profile.display_name),
-                                 self.buff_type.pack("?", False),  # No signature as we haven't implemented them here
-                                 self.buff_type.pack("?", False))  # No UUID as we haven't implemented them yet
+                                 *signature,
+                                 *uuid)
             elif self.protocol_version == 759:  # 1.19
                 self.send_packet("login_start",
                                  self.buff_type.pack_string(self.factory.profile.display_name),
-                                 self.buff_type.pack("?", False))  # No signature as we haven't implemented them here
+                                 *signature)  # No signature as we haven't implemented them here
             else:
                 # Send login start
                 self.send_packet("login_start", self.buff_type.pack_string(
@@ -98,13 +128,60 @@ def auth_ok(self, data):
             pack_array = lambda d: self.buff_type.pack_varint(
                 len(d), max_bits=16) + d
 
+        # Notes on 1.19+ encryption (from wiki.vg, testing, and Yarn mappings of a 1.19.2 Fabric server)
+        # Mapping help: https://linkie.shedaniel.me/mappings?namespace=yarn&version=1.19.2&translateAs=mojang&search=
+        #
+        # read_either: one bit (true/false) to indicate which of the following two fields is present
+        # token case:
+        # + first bit is 1
+        # + send over pre-encrypted p_verify_token as-is
+        #
+        # signature case:
+        # - first bit is 0
+        # - vanilla server calls SignatureData::new (Yarn 1.19.2)
+        # - ordering in packet: readLong -> salt; readByteArray -> signature
+        # - !! Don't sign the encrypted p_verify_token, sign the UNENCRYPTED self.verify_token !!
+        # - 1. generate random 8 bytes ("salt" or "seed" depending on context)
+        # - 2. sign the verify_token with the private key, specifically verify_token concat with salt
+        # - 3. send the signed verify_token + salt (bytes) back
+
         # 1.19+
         if self.protocol_version >= 759:
+            profile: Profile = self.factory.profile
+            can_use_signing_data = (
+                    profile.online
+                    and isinstance(profile, Profile)
+                    and profile.enable_signing
+                    and profile.certificates is not None
+            )
+            # true = old way; false = new way
+            verify = [self.buff_type.pack("?", not can_use_signing_data)]
+            if not can_use_signing_data:
+                verify += [pack_array(p_verify_token)]
+            else:
+                salt = bytearray(os.urandom(8))
+                nonce = self.verify_token  # NOT p_verify_token!!
+                salt_number = int.from_bytes(salt, "big")
+
+                # Can't + bytes to concatenate, temporarily switch to bytearray
+                what_to_sign = bytearray(nonce) + salt
+                key = ssl_crypto.load_privatekey(
+                    ssl_crypto.FILETYPE_PEM,
+                    profile.certificates.private
+                )
+                signature = ssl_crypto.sign(
+                    key,
+                    bytes(what_to_sign),  # Switch back to bytes for signing
+                    "sha256"
+                )
+                verify += [
+                    self.buff_type.pack('Q', salt_number),
+                    self.buff_type.pack_byte_array(signature)
+                ]
             self.send_packet(
                 "login_encryption_response",
                 pack_array(p_shared_secret),
-                self.buff_type.pack('?', True),  # Indicate we are still doing things the old way
-                pack_array(p_verify_token))
+                *verify)
         else:
             self.send_packet(
                 "login_encryption_response",

quarry/net/http.py

@@ -35,7 +35,7 @@ def stopProducing(self):
         pass
 
 
-def request(url, timeout, err_type=Exception, expect_content=False, data=None):
+def request(url, timeout, err_type=Exception, expect_content=False, data=None, headers=None):
     d0 = defer.Deferred()
 
     def _callback(response):
@@ -67,15 +67,19 @@ def _errback(err):
 
     agent = Agent(reactor)
 
+    if headers is None:
+        headers = {}
+
     if data:
+        headers.update({"Content-Type": ["application/json"]})
         d1 = agent.request(
             b'POST',
             url,
-            Headers({"Content-Type": ["application/json"]}),
+            Headers(headers),
             BytesProducer(json.dumps(data).encode('ascii')),
         )
     else:
-        d1 = agent.request(b'GET', url)
+        d1 = agent.request(b'GET', url, Headers(headers))
 
     d1.addCallbacks(_callback, _errback)
 

quarry/types/certificate.py

@@ -0,0 +1,105 @@
+import base64
+import re
+import typing
+from datetime import datetime
+
+
+def str_bytes(string) -> bytes:
+    if isinstance(string, str):
+        return bytes(string, encoding="UTF-8")
+    elif isinstance(string, bytes):
+        return string
+    raise ValueError(f"str_bytes expected str or bytes, got {type(string)}")
+
+
+def bytes_str(byte_s) -> str:
+    if isinstance(byte_s, str):
+        return byte_s
+    elif isinstance(byte_s, bytes):
+        return byte_s.decode(encoding="UTF-8")
+    raise ValueError(f"bytes_str expected str or bytes, got {type(byte_s)}")
+
+
+class CertificatePair(object):
+    BytesOrStr = typing.Union[str, bytes]
+
+    def __init__(
+        self,
+        private: BytesOrStr,
+        public: BytesOrStr,
+        signature1: BytesOrStr,
+        signature2: BytesOrStr,
+        expires: str
+    ):
+        self.private = str_bytes(private)
+        self.public = str_bytes(public)
+        self.signature1 = str_bytes(signature1)
+        self.signature2 = str_bytes(signature2)
+
+        # ISO 8601, datetime really doesn't like extra fractional second precision
+
+        # Capture up to 6 digits of fractional seconds, or none at all
+        expires = re.sub(r'^(\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d{1,6})?)\d+(Z|\+[\d:]+)$', r'\1\2', expires)
+        self.expires = datetime.strptime(expires, '%Y-%m-%dT%H:%M:%S.%f%z')  # cpython issue #80010
+
+    def is_expired(self):
+        return datetime.now() > self.expires
+
+    def __eq__(self, other):
+        if not isinstance(other, CertificatePair):
+            return False
+        return (
+            self.private == other.private
+            and self.public == other.public
+            and self.signature1 == other.signature1
+            and self.signature2 == other.signature2
+            and self.expires == other.expires
+        )
+
+    def __lt__(self, other):
+        if not isinstance(other, CertificatePair):
+            return False
+        return self.expires < other.expires
+
+    def __gt__(self, other):
+        if not isinstance(other, CertificatePair):
+            return False
+        return self.expires > other.expires
+
+    def __le__(self, other):
+        if not isinstance(other, CertificatePair):
+            return False
+        return self.expires <= other.expires
+
+    def __ge__(self, other):
+        if not isinstance(other, CertificatePair):
+            return False
+        return self.expires >= other.expires
+
+    def __repr__(self):
+        return f"{type(self).__name__}(pub={self.public[:16]}, s1={self.signature1[:16]}, s2={self.signature2[:16]}, expire={self.expires.isoformat()})"
+
+    @classmethod
+    def convert_public_key(cls, cert_pem):
+
+        contents = r'-{3,}BEGIN[ A-Z]*-{3,}\n?((?:[a-zA-Z0-9/+]*\n?)*)-{3,}END[ A-Z]*-{3,}'
+        contents = re.search(contents, bytes_str(cert_pem))
+        if contents is None:
+            raise ValueError("Invalid certificate (failed to parse)")
+        contents = contents.group(1)
+        contents = re.sub(r'\s', '', contents)
+        ba = base64.b64decode(contents)
+        return ba
+
+    @classmethod
+    def from_dict(cls, certificates):
+        return cls(**certificates)
+
+    def to_dict(self):
+        return {
+            "private": self.private,
+            "public": self.public,
+            "signature1": self.signature1,
+            "signature2": self.signature2,
+            "expires": self.expires.isoformat()
+        }