1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
|
"""
Signed Session
A wrapper around httpx' AsyncClient which transparently calculates the "Signature" header.
"""
import httpx
from xbox.webapi.common.request_signer import RequestSigner
class SignedSession(httpx.AsyncClient):
def __init__(self, request_signer=None):
super().__init__()
self.request_signer = request_signer or RequestSigner()
@classmethod
def from_pem_signing_key(cls, pem_string: str):
request_signer = RequestSigner.from_pem(pem_string)
return cls(request_signer)
def _prepare_signed_request(self, request: httpx.Request) -> httpx.Request:
path_and_query = request.url.raw_path.decode()
authorization = request.headers.get("Authorization", "")
body = b""
for byte in request.stream:
body += byte
signature = self.request_signer.sign(
method=request.method,
path_and_query=path_and_query,
body=body,
authorization=authorization,
)
request.headers["Signature"] = signature
return request
async def send_request_signed(self, request: httpx.Request) -> httpx.Response:
"""
Shorthand for prepare signed + send
"""
prepared = self._prepare_signed_request(request)
return await self.send(prepared)
async def send_signed(self, method: str, url: str, **kwargs):
"""
Shorthand for creating request + prepare signed + send
"""
request = httpx.Request(method, url, **kwargs)
prepared = self._prepare_signed_request(request)
return await self.send(prepared)
|
