Merge branch 'feat/xal_auth'

Author: tuxuser

setup.py

@@ -65,6 +65,7 @@
     entry_points={
         "console_scripts": [
             "xbox-authenticate=xbox.webapi.scripts.authenticate:main",
+            "xbox-xal=xbox.webapi.scripts.xal:main",
             "xbox-searchlive=xbox.webapi.scripts.search:main",
             "xbox-change-gt=xbox.webapi.scripts.change_gamertag:main",
             "xbox-friends=xbox.webapi.scripts.friends:main",

tests/conftest.py

@@ -1,4 +1,5 @@
 from datetime import datetime
+import uuid
 
 from ecdsa.keys import SigningKey, VerifyingKey
 import pytest
@@ -11,6 +12,11 @@
     XAUResponse,
     XSTSResponse,
 )
+from xbox.webapi.authentication.xal import (
+    APP_PARAMS_GAMEPASS_BETA,
+    CLIENT_PARAMS_ANDROID,
+    XALManager,
+)
 from xbox.webapi.common.request_signer import RequestSigner
 from xbox.webapi.common.signed_session import SignedSession
 
@@ -30,6 +36,19 @@ async def auth_mgr(event_loop):
     await session.aclose()
 
 
+@pytest_asyncio.fixture(scope="function")
+async def xal_mgr(event_loop):
+    session = SignedSession()
+    mgr = XALManager(
+        session,
+        device_id=uuid.UUID("9c493431-5462-4a4a-a247-f6420396318d"),
+        app_params=APP_PARAMS_GAMEPASS_BETA,
+        client_params=CLIENT_PARAMS_ANDROID,
+    )
+    yield mgr
+    await session.aclose()
+
+
 @pytest.fixture(scope="function")
 def xbl_client(auth_mgr):
     return XboxLiveClient(auth_mgr)

tests/data/responses/xal_authentication_resp.json

@@ -0,0 +1,4 @@
+{
+    "MsaOauthRedirect":"https://login.live.com/oauth20_authorize.srf?lw=1&fl=dob,easi2&xsup=1&display=android_phone&code_challenge=code_challenge_string&code_challenge_method=S256&state=state_string&client_id=000000004C20A908&response_type=code&scope=service%3A%3Auser.auth.xboxlive.com%3A%3AMBI_SSL&redirect_uri=ms-xal-public-beta-000000004c20a908%3A%2F%2Fauth&nopa=2",
+    "MsaRequestParameters":{}
+}

tests/data/responses/xal_authorization_resp.json

@@ -0,0 +1,44 @@
+{
+    "DeviceToken": "eyDeviceToken",
+    "TitleToken": {
+        "DisplayClaims": {
+            "xti": {
+                "tid": "1016898439"
+            }
+        },
+        "IssueInstant": "2022-11-11T21:11:43.9456623Z",
+        "NotAfter": "2099-11-25T21:11:43.9456623Z",
+        "Token": "eyTitletoken"
+    },
+    "UserToken": {
+        "DisplayClaims": {
+            "xui": [
+                {
+                    "uhs": "2034583485034500345"
+                }
+            ]
+        },
+        "IssueInstant": "2022-11-11T21:11:43.9422756Z",
+        "NotAfter": "2099-11-25T21:11:43.9422756Z",
+        "Token": "eyUserToken"
+    },
+    "AuthorizationToken": {
+        "DisplayClaims": {
+            "xui": [
+                {
+                    "gtg": "Pony",
+                    "xid": "24812480912094",
+                    "uhs": "2034583485034500345",
+                    "agg": "Adult",
+                    "usr": "195 229 243",
+                    "prv": "184 185 186 187 188 190 191 192 193 194 196 198 199 200 201 203 204 205 206 207 208 211 214 215 216 217 220 224 227 228 235 238 245 247 249 252 254 255"
+                }
+            ]
+        },
+        "IssueInstant": "2022-11-11T21:11:44.1945885Z",
+        "NotAfter": "2099-11-12T13:11:44.1945885Z",
+        "Token": "eyXSTSToken"
+    },
+    "WebPage": "https://sisu.xboxlive.com/client/v27/000000004c20a908/view/index.html",
+    "Sandbox": "RETAIL"
+}

tests/test_auth.py

@@ -1,5 +1,4 @@
 from datetime import datetime, timedelta, timezone
-import uuid
 
 from httpx import Response
 import pytest
@@ -90,26 +89,6 @@ async def test_refresh_tokens_user_still_valid(respx_mock, auth_mgr):
     assert route2.called
 
 
-@pytest.mark.asyncio
-async def test_get_title_endpoints(respx_mock, auth_mgr):
-    route = respx_mock.get("https://title.mgt.xboxlive.com").mock(
-        return_value=Response(200, json=get_response_json("auth_title_endpoints"))
-    )
-    await auth_mgr.get_title_endpoints()
-    assert route.called
-
-
-@pytest.mark.asyncio
-async def test_get_device_token(respx_mock, auth_mgr):
-    route = respx_mock.post(
-        "https://device.auth.xboxlive.com/device/authenticate"
-    ).mock(return_value=Response(200, json=get_response_json("auth_device_token")))
-    resp = await auth_mgr.request_device_token(
-        uuid.UUID("9c493431-5462-4a4a-a247-f6420396318d")
-    )
-    assert route.called
-
-
 @pytest.mark.asyncio
 async def test_xsts_properties(auth_mgr):
     assert auth_mgr.xsts_token.xuid == "2669321029139235"

tests/test_signed_session.py

@@ -29,7 +29,34 @@ async def test_sending_signed_request(synthetic_request_signer, respx_mock):
     )
 
     async with signed_session:
-        resp = await signed_session.send_signed(request)
+        resp = await signed_session.send_request_signed(request)
+
+    assert route.called
+    assert resp.request.headers.get("Signature") is not None
+
+
+@pytest.mark.asyncio
+async def test_sending_signed(synthetic_request_signer, respx_mock):
+    route = respx_mock.post("https://xsts.auth.xboxlive.com").mock(
+        return_value=Response(200, json=get_response_json("auth_xsts_token"))
+    )
+
+    signed_session = SignedSession(synthetic_request_signer)
+
+    method = "POST"
+    url = "https://xsts.auth.xboxlive.com/xsts/authorize"
+    headers = {"x-xbl-contract-version": "1"}
+    data = {
+        "RelyingParty": "http://xboxlive.com",
+        "TokenType": "JWT",
+        "Properties": {
+            "UserTokens": ["eyJWTblabla"],
+            "SandboxId": "RETAIL",
+        },
+    }
+
+    async with signed_session:
+        resp = await signed_session.send_signed(method, url, headers=headers, data=data)
 
     assert route.called
     assert resp.request.headers.get("Signature") is not None

tests/test_xal.py

@@ -0,0 +1,63 @@
+import uuid
+
+from httpx import AsyncClient, Response
+import pytest
+
+from tests.common import get_response_json
+
+
+@pytest.mark.asyncio
+async def test_get_title_endpoints(respx_mock, xal_mgr):
+    route = respx_mock.get("https://title.mgt.xboxlive.com").mock(
+        return_value=Response(200, json=get_response_json("auth_title_endpoints"))
+    )
+    async with AsyncClient() as client:
+        await xal_mgr.get_title_endpoints(client)
+    assert route.called
+
+
+@pytest.mark.asyncio
+async def test_get_device_token(respx_mock, xal_mgr):
+    route = respx_mock.post(
+        "https://device.auth.xboxlive.com/device/authenticate"
+    ).mock(return_value=Response(200, json=get_response_json("auth_device_token")))
+    await xal_mgr.request_device_token()
+    assert route.called
+
+
+@pytest.mark.asyncio
+async def test_sisu_authentication(respx_mock, xal_mgr):
+    route = respx_mock.post("https://sisu.xboxlive.com/authenticate").mock(
+        return_value=Response(
+            200,
+            json=get_response_json("xal_authentication_resp"),
+            headers={"X-SessionId": "abcsession-id"},
+        )
+    )
+    resp, session_id = await xal_mgr.request_sisu_authentication(
+        "eyDeviceToken", "code_challenge_string", "state_string"
+    )
+    assert route.called
+    assert session_id == "abcsession-id"
+    assert resp.msa_oauth_redirect is not None
+
+
+@pytest.mark.asyncio
+async def test_sisu_authorization(respx_mock, xal_mgr):
+    route = respx_mock.post("https://sisu.xboxlive.com/authorize").mock(
+        return_value=Response(200, json=get_response_json("xal_authorization_resp"))
+    )
+    await xal_mgr.do_sisu_authorization(
+        "SISU-Session-ID", "eyAccessToken", "eyDeviceToken"
+    )
+    assert route.called
+
+
+@pytest.mark.asyncio
+async def test_exchange_code_for_token(respx_mock, xal_mgr):
+    route = respx_mock.post("https://login.live.com").mock(
+        return_value=Response(200, json=get_response_json("auth_oauth2_token"))
+    )
+    await xal_mgr.exchange_code_for_token("abc", "xyz")
+
+    assert route.called

xbox/webapi/authentication/manager.py

@@ -5,14 +5,11 @@
 """
 import logging
 from typing import List, Optional
-import uuid
 
 import httpx
 
 from xbox.webapi.authentication.models import (
     OAuth2TokenResponse,
-    TitleEndpointsResponse,
-    XADResponse,
     XAUResponse,
     XSTSResponse,
 )
@@ -83,14 +80,6 @@ async def refresh_tokens(self) -> None:
         if not (self.xsts_token and self.xsts_token.is_valid()):
             self.xsts_token = await self.request_xsts_token()
 
-    async def get_title_endpoints(self) -> TitleEndpointsResponse:
-        url = "https://title.mgt.xboxlive.com/titles/default/endpoints"
-        headers = {"x-xbl-contract-version": "1"}
-        params = {"type": 1}
-        resp = await self.session.get(url, headers=headers, params=params)
-        resp.raise_for_status()
-        return TitleEndpointsResponse(**resp.json())
-
     async def request_oauth_token(self, authorization_code: str) -> OAuth2TokenResponse:
         """Request OAuth2 token."""
         return await self._oauth2_token_request(
@@ -170,23 +159,3 @@ async def request_xsts_token(
             raise AuthenticationException()
         resp.raise_for_status()
         return XSTSResponse(**resp.json())
-
-    async def request_device_token(self, device_id: uuid.UUID) -> XADResponse:
-        url = "https://device.auth.xboxlive.com/device/authenticate"
-        headers = {"x-xbl-contract-version": "1"}
-        data = {
-            "RelyingParty": "http://auth.xboxlive.com",
-            "TokenType": "JWT",
-            "Properties": {
-                "AuthMethod": "ProofOfPossession",
-                "Id": str(device_id).upper(),
-                "DeviceType": "Win32",
-                "Version": "10.0.22000.194",
-                "ProofKey": self.session.request_signer.proof_field,
-            },
-        }
-
-        request = httpx.Request("POST", url, headers=headers, json=data)
-        resp = await self.session.send_signed(request)
-        resp.raise_for_status()
-        return XADResponse(**resp.json())

xbox/webapi/authentication/models.py

@@ -3,6 +3,7 @@
 from typing import Dict, List, Optional
 
 from pydantic import BaseModel, Field
+from pydantic.dataclasses import dataclass
 
 from xbox.webapi.common.models import PascalCaseModel
 
@@ -94,6 +95,39 @@ def is_valid(self) -> bool:
         return (self.issued + timedelta(seconds=self.expires_in)) > utc_now()
 
 
+"""XAL related models"""
+
+
+@dataclass
+class XalAppParameters:
+    app_id: str
+    title_id: str
+    redirect_uri: str
+
+
+@dataclass
+class XalClientParameters:
+    user_agent: str
+    device_type: str
+    client_version: str
+    query_display: str
+
+
+class SisuAuthenticationResponse(PascalCaseModel):
+    msa_oauth_redirect: str
+    msa_request_parameters: Dict[str, str]
+
+
+class SisuAuthorizationResponse(PascalCaseModel):
+    device_token: str
+    title_token: XATResponse
+    user_token: XAUResponse
+    authorization_token: XSTSResponse
+    web_page: str
+    sandbox: str
+    use_modern_gamertag: Optional[bool]
+
+
 """Signature related models"""