signed session: Add example of usage to AuthenticationManager

Author: tuxuser

README.md

@@ -59,7 +59,8 @@ API usage
 ```py
 import sys
 import asyncio
-from httpx import AsyncClient, HTTPStatusError
+from httpx import HTTPStatusError
+from xbox.webapi.common.signed_session import SignedSession
 from xbox.webapi.api.client import XboxLiveClient
 from xbox.webapi.authentication.manager import AuthenticationManager
 from xbox.webapi.authentication.models import OAuth2TokenResponse
@@ -75,7 +76,7 @@ tokens_file = TOKENS_FILE
 For doing authentication, see xbox/webapi/scripts/authenticate.py
 """
 async def async_main():
-    async with AsyncClient() as session:
+    async with SignedSession() as session:
         auth_mgr = AuthenticationManager(
               session, client_id, client_secret, ""
         )

readme_example.py

@@ -1,9 +1,12 @@
-import sys
 import asyncio
-from httpx import AsyncClient, HTTPStatusError
+import sys
+
+from httpx import HTTPStatusError
+
 from xbox.webapi.api.client import XboxLiveClient
 from xbox.webapi.authentication.manager import AuthenticationManager
 from xbox.webapi.authentication.models import OAuth2TokenResponse
+from xbox.webapi.common.signed_session import SignedSession
 from xbox.webapi.scripts import CLIENT_ID, CLIENT_SECRET, TOKENS_FILE
 
 # This uses the default client identification by OpenXbox
@@ -15,55 +18,58 @@
 """
 For doing authentication, see xbox/webapi/scripts/authenticate.py
 """
+
+
 async def async_main():
-    async with AsyncClient() as session:
-        auth_mgr = AuthenticationManager(
-              session, client_id, client_secret, ""
-        )
+    async with SignedSession() as session:
+        auth_mgr = AuthenticationManager(session, client_id, client_secret, "")
 
         try:
-            with open(tokens_file, mode="r") as f:
-                  tokens = f.read()
+            with open(tokens_file) as f:
+                tokens = f.read()
             auth_mgr.oauth = OAuth2TokenResponse.parse_raw(tokens)
         except FileNotFoundError:
-            print(f'File {tokens_file} isn`t found or it doesn`t contain tokens!')
+            print(f"File {tokens_file} isn`t found or it doesn`t contain tokens!")
             exit(-1)
-        
+
         try:
-              await auth_mgr.refresh_tokens()
+            await auth_mgr.refresh_tokens()
         except HTTPStatusError:
-              print("Could not refresh tokens")
-              sys.exit(-1)
+            print("Could not refresh tokens")
+            sys.exit(-1)
 
         with open(tokens_file, mode="w") as f:
-              f.write(auth_mgr.oauth.json())
-        print(f'Refreshed tokens in {tokens_file}!')
+            f.write(auth_mgr.oauth.json())
+        print(f"Refreshed tokens in {tokens_file}!")
 
         xbl_client = XboxLiveClient(auth_mgr)
 
         # Some example API calls
         # Get friendslist
         friendslist = await xbl_client.people.get_friends_own()
-        print('Your friends:')
+        print("Your friends:")
         print(friendslist)
         print()
 
         # Get presence status (by list of XUID)
-        presence = await xbl_client.presence.get_presence_batch(["2533274794093122", "2533274807551369"])
-        print('Statuses of some random players by XUID:')
+        presence = await xbl_client.presence.get_presence_batch(
+            ["2533274794093122", "2533274807551369"]
+        )
+        print("Statuses of some random players by XUID:")
         print(presence)
         print()
 
         # Get messages
         messages = await xbl_client.message.get_inbox()
-        print('Your messages:')
+        print("Your messages:")
         print(messages)
         print()
 
         # Get profile by GT
         profile = await xbl_client.profile.get_profile_by_gamertag("SomeGamertag")
-        print('Profile under SomeGamertag gamer tag:')
+        print("Profile under SomeGamertag gamer tag:")
         print(profile)
         print()
 
-asyncio.run(async_main())
+
+asyncio.run(async_main())

tests/conftest.py

@@ -1,7 +1,6 @@
 from datetime import datetime
 
 from ecdsa.keys import SigningKey, VerifyingKey
-from httpx import AsyncClient
 import pytest
 import pytest_asyncio
 
@@ -13,6 +12,7 @@
     XSTSResponse,
 )
 from xbox.webapi.common.request_signer import RequestSigner
+from xbox.webapi.common.signed_session import SignedSession
 
 from tests.common import get_response
 
@@ -21,7 +21,7 @@
 
 @pytest_asyncio.fixture(scope="function")
 async def auth_mgr(event_loop):
-    session = AsyncClient()
+    session = SignedSession()
     mgr = AuthenticationManager(session, "abc", "123", "http://localhost")
     mgr.oauth = OAuth2TokenResponse.parse_raw(get_response("auth_oauth2_token"))
     mgr.user_token = XAUResponse.parse_raw(get_response("auth_user_token"))
@@ -45,10 +45,12 @@ def ecdsa_signing_key_str() -> str:
 def ecdsa_signing_key(ecdsa_signing_key_str: str) -> SigningKey:
     return SigningKey.from_pem(ecdsa_signing_key_str)
 
+
 @pytest.fixture(scope="session")
 def ecdsa_verifying_key(ecdsa_signing_key: SigningKey) -> VerifyingKey:
     return ecdsa_signing_key.get_verifying_key()
 
+
 @pytest.fixture(scope="session")
 def synthetic_request_signer(ecdsa_signing_key) -> RequestSigner:
     return RequestSigner(ecdsa_signing_key)

tests/data/responses/auth_device_token.json

@@ -0,0 +1,11 @@
+{
+    "IssueInstant": "2010-10-10T03:04:29.6037497Z",
+    "NotAfter": "2999-10-24T03:04:29.6037497Z",
+    "Token": "eyJhYoToken",
+    "DisplayClaims": {
+        "xdi": {
+            "did": "F9E13DC7B0997D0D",
+            "dcs": "0"
+        }
+    }
+}

tests/test_auth.py

@@ -1,4 +1,5 @@
 from datetime import datetime, timedelta, timezone
+import uuid
 
 from httpx import Response
 import pytest
@@ -60,7 +61,7 @@ async def test_refresh_tokens(respx_mock, auth_mgr):
 
 
 @pytest.mark.asyncio
-async def test_refresh_tokens_still_valid(respx_mock, auth_mgr):
+async def test_refresh_tokens_still_valid(auth_mgr):
     now = datetime.now(timezone.utc)
     auth_mgr.oauth.issued = now
     auth_mgr.user_token.not_after = now + timedelta(days=1)
@@ -96,6 +97,17 @@ async def test_get_title_endpoints(respx_mock, auth_mgr):
     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"

xbox/webapi/authentication/manager.py

@@ -5,17 +5,20 @@
 """
 import logging
 from typing import List, Optional
+import uuid
 
 import httpx
 from yarl import URL
 
 from xbox.webapi.authentication.models import (
     OAuth2TokenResponse,
     TitleEndpointsResponse,
+    XADResponse,
     XAUResponse,
     XSTSResponse,
 )
 from xbox.webapi.common.exceptions import AuthenticationException
+from xbox.webapi.common.signed_session import SignedSession
 
 log = logging.getLogger("authentication")
 
@@ -25,13 +28,13 @@
 class AuthenticationManager:
     def __init__(
         self,
-        client_session: httpx.Client,
+        client_session: SignedSession,
         client_id: str,
         client_secret: str,
         redirect_uri: str,
         scopes: Optional[List[str]] = None,
     ):
-        self.session: httpx.Client = client_session
+        self.session: SignedSession = client_session
         self._client_id: str = client_id
         self._client_secret: str = client_secret
         self._redirect_uri: str = redirect_uri
@@ -160,3 +163,23 @@ 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/scripts/authenticate.py

@@ -11,10 +11,9 @@
 from urllib.parse import parse_qs, urlparse
 import webbrowser
 
-from httpx import AsyncClient
-
 from xbox.webapi.authentication.manager import AuthenticationManager
 from xbox.webapi.authentication.models import OAuth2TokenResponse
+from xbox.webapi.common.signed_session import SignedSession
 from xbox.webapi.scripts import CLIENT_ID, CLIENT_SECRET, REDIRECT_URI, TOKENS_FILE
 
 QUEUE = queue.Queue(1)
@@ -74,14 +73,14 @@ def do_GET(self):
 async def do_auth(
     client_id: str, client_secret: str, redirect_uri: str, token_filepath: str
 ):
-    async with AsyncClient() as session:
+    async with SignedSession() as session:
         auth_mgr = AuthenticationManager(
             session, client_id, client_secret, redirect_uri
         )
 
         # Refresh tokens if we have them
         if os.path.exists(token_filepath):
-            with open(token_filepath, mode="r") as f:
+            with open(token_filepath) as f:
                 tokens = f.read()
             auth_mgr.oauth = OAuth2TokenResponse.parse_raw(tokens)
             await auth_mgr.refresh_tokens()

xbox/webapi/scripts/change_gamertag.py

@@ -6,7 +6,7 @@
 import os
 import sys
 
-from httpx import AsyncClient, HTTPStatusError
+from httpx import HTTPStatusError
 
 from xbox.webapi.api.client import XboxLiveClient
 from xbox.webapi.api.provider.account.models import (
@@ -15,6 +15,7 @@
 )
 from xbox.webapi.authentication.manager import AuthenticationManager
 from xbox.webapi.authentication.models import OAuth2TokenResponse
+from xbox.webapi.common.signed_session import SignedSession
 from xbox.webapi.scripts import CLIENT_ID, CLIENT_SECRET, TOKENS_FILE
 
 
@@ -50,12 +51,12 @@ async def async_main():
         print("No token file found, run xbox-authenticate")
         sys.exit(-1)
 
-    async with AsyncClient() as session:
+    async with SignedSession() as session:
         auth_mgr = AuthenticationManager(
             session, args.client_id, args.client_secret, ""
         )
 
-        with open(args.tokens, mode="r") as f:
+        with open(args.tokens) as f:
             tokens = f.read()
         auth_mgr.oauth = OAuth2TokenResponse.parse_raw(tokens)
         try:

xbox/webapi/scripts/friends.py

@@ -7,11 +7,12 @@
 from pprint import pprint
 import sys
 
-from httpx import AsyncClient, HTTPStatusError
+from httpx import HTTPStatusError
 
 from xbox.webapi.api.client import XboxLiveClient
 from xbox.webapi.authentication.manager import AuthenticationManager
 from xbox.webapi.authentication.models import OAuth2TokenResponse
+from xbox.webapi.common.signed_session import SignedSession
 from xbox.webapi.scripts import CLIENT_ID, CLIENT_SECRET, TOKENS_FILE
 
 
@@ -42,12 +43,12 @@ async def async_main():
         print("No token file found, run xbox-authenticate")
         sys.exit(-1)
 
-    async with AsyncClient() as session:
+    async with SignedSession() as session:
         auth_mgr = AuthenticationManager(
             session, args.client_id, args.client_secret, ""
         )
 
-        with open(args.tokens, mode="r") as f:
+        with open(args.tokens) as f:
             tokens = f.read()
         auth_mgr.oauth = OAuth2TokenResponse.parse_raw(tokens)
         try:

xbox/webapi/scripts/search.py

@@ -6,10 +6,11 @@
 from pprint import pprint
 import sys
 
-from httpx import AsyncClient, HTTPStatusError
+from httpx import HTTPStatusError
 
 from xbox.webapi.api.client import XboxLiveClient
 from xbox.webapi.authentication.manager import AuthenticationManager
+from xbox.webapi.common.signed_session import SignedSession
 
 
 async def async_main():
@@ -18,7 +19,7 @@ async def async_main():
 
     args = parser.parse_args()
 
-    async with AsyncClient() as session:
+    async with SignedSession() as session:
         auth_mgr = AuthenticationManager(session, "", "", "")
 
         # No Auth necessary for catalog searches