Skip to content

bovine.clients.signed_http

SignedHttpClient

Client for using HTTP Signatures

Source code in bovine/bovine/clients/signed_http.py 
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
class SignedHttpClient:
    """Client for using HTTP Signatures"""

    def __init__(
        self,
        session: aiohttp.ClientSession,
        public_key_url: str,
        private_key: str,
        digest_method: Callable[[bytes], Tuple[str, str]] | None = None,
    ):
        """Creates the http client. By using [content_digest_sha256_rfc_9530][bovine.crypto.helper.content_digest_sha256_rfc_9530] one can send the Content Digest according to RFC 9530.

        :param session: The aiohttp.ClientSession
        :param public_key_url: Used as keyId when signing
        :param private_key: The pem encoded private key
        :param digest_method: Allow specifying the method used to compute the digest
        """
        self.session = session
        self.secret = CryptographicSecret.from_pem(public_key_url, private_key)

        if digest_method is None:
            digest_method = legacy_digest_method
        self.digest_method = digest_method

    async def get(self, url: str, headers: dict = {}) -> aiohttp.ClientResponse:
        """Retrieves url using a signed get request

        :param url: URL to get
        :param headers: extra headers"""
        return await bovine.clients.signed_http_methods.signed_get(
            self.session, self.secret, url, headers
        )

    async def post(
        self, url: str, body: str, headers: dict = {}, content_type: str = None
    ):
        """Posts to url using a signed post request

        :param url: URL to post to
        :param body: The post body
        :param headers: extra headers
        :param content_type: Content type of the message"""
        return await bovine.clients.signed_http_methods.signed_post(
            self.session,
            self.secret,
            url,
            body,
            headers=headers,
            digest_method=self.digest_method,
            content_type=content_type,
        )

    def event_source(self, url: str):
        """Opens an event source to url

        :param url: Url to query"""
        return bovine.clients.signed_http_methods.signed_event_source(
            self.session, self.secret, url
        )

__init__ 

__init__(
    session: ClientSession,
    public_key_url: str,
    private_key: str,
    digest_method: (
        Callable[[bytes], Tuple[str, str]] | None
    ) = None,
)

Creates the http client. By using content_digest_sha256_rfc_9530 one can send the Content Digest according to RFC 9530.

Parameters:

Name Type Description Default
session ClientSession

The aiohttp.ClientSession

 required
public_key_url str

Used as keyId when signing

 required
private_key str

The pem encoded private key

 required
digest_method Callable[[bytes], Tuple[str, str]] | None

Allow specifying the method used to compute the digest

 None
Source code in bovine/bovine/clients/signed_http.py 
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
def __init__(
    self,
    session: aiohttp.ClientSession,
    public_key_url: str,
    private_key: str,
    digest_method: Callable[[bytes], Tuple[str, str]] | None = None,
):
    """Creates the http client. By using [content_digest_sha256_rfc_9530][bovine.crypto.helper.content_digest_sha256_rfc_9530] one can send the Content Digest according to RFC 9530.

    :param session: The aiohttp.ClientSession
    :param public_key_url: Used as keyId when signing
    :param private_key: The pem encoded private key
    :param digest_method: Allow specifying the method used to compute the digest
    """
    self.session = session
    self.secret = CryptographicSecret.from_pem(public_key_url, private_key)

    if digest_method is None:
        digest_method = legacy_digest_method
    self.digest_method = digest_method

event_source 

event_source(url: str)

Opens an event source to url

Parameters:

Name Type Description Default
url str

Url to query

 required
Source code in bovine/bovine/clients/signed_http.py 
90
91
92
93
94
95
96
def event_source(self, url: str):
    """Opens an event source to url

    :param url: Url to query"""
    return bovine.clients.signed_http_methods.signed_event_source(
        self.session, self.secret, url
    )

get async 

get(url: str, headers: dict = {}) -> ClientResponse

Retrieves url using a signed get request

Parameters:

Name Type Description Default
url str

URL to get

 required
headers dict

extra headers

 {}
Source code in bovine/bovine/clients/signed_http.py 
62
63
64
65
66
67
68
69
async def get(self, url: str, headers: dict = {}) -> aiohttp.ClientResponse:
    """Retrieves url using a signed get request

    :param url: URL to get
    :param headers: extra headers"""
    return await bovine.clients.signed_http_methods.signed_get(
        self.session, self.secret, url, headers
    )

post async 

post(
    url: str,
    body: str,
    headers: dict = {},
    content_type: str = None,
)

Posts to url using a signed post request

Parameters:

Name Type Description Default
url str

URL to post to

 required
body str

The post body

 required
headers dict

extra headers

 {}
content_type str

Content type of the message

 None
Source code in bovine/bovine/clients/signed_http.py 
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
async def post(
    self, url: str, body: str, headers: dict = {}, content_type: str = None
):
    """Posts to url using a signed post request

    :param url: URL to post to
    :param body: The post body
    :param headers: extra headers
    :param content_type: Content type of the message"""
    return await bovine.clients.signed_http_methods.signed_post(
        self.session,
        self.secret,
        url,
        body,
        headers=headers,
        digest_method=self.digest_method,
        content_type=content_type,
    )

legacy_digest_method 

legacy_digest_method(body)

Digest method with sha-256 as prefix

>>> legacy_digest_method("moo")
('digest', 'sha-256=R9+ukoir89XSJSq/sL1qyWYmN9ZG5t+dXSdLwzbierw=')
Source code in bovine/bovine/clients/signed_http.py 
14
15
16
17
18
19
20
21
22
23
def legacy_digest_method(body):
    """Digest method with sha-256 as prefix

    ```pycon
    >>> legacy_digest_method("moo")
    ('digest', 'sha-256=R9+ukoir89XSJSq/sL1qyWYmN9ZG5t+dXSdLwzbierw=')

    ```
    """
    return "digest", content_digest_sha256(body)

legacy_digest_method_uppercase 

legacy_digest_method_uppercase(body)

Digest method with SHA-256 as prefix

>>> legacy_digest_method_uppercase("moo")
('digest', 'SHA-256=R9+ukoir89XSJSq/sL1qyWYmN9ZG5t+dXSdLwzbierw=')
Source code in bovine/bovine/clients/signed_http.py 
26
27
28
29
30
31
32
33
34
35
def legacy_digest_method_uppercase(body):
    """Digest method with SHA-256 as prefix

    ```pycon
    >>> legacy_digest_method_uppercase("moo")
    ('digest', 'SHA-256=R9+ukoir89XSJSq/sL1qyWYmN9ZG5t+dXSdLwzbierw=')

    ```
    """
    return "digest", content_digest_sha256(body, prefix="SHA-256")