Skip to content

bovine.clients.moo_auth

MooAuthClient dataclass

Client for using Moo-Auth-1 authentication

Parameters:

Name Type Description Default
session ClientSession

The session

 required
did_key str

The did key, i.e. did:key:z...

 required
private_key str

private key corresponding to did_key

 required
Source code in bovine/bovine/clients/moo_auth.py 
 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
 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
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
@dataclass
class MooAuthClient:
    """Client for using Moo-Auth-1 authentication"""

    session: aiohttp.ClientSession = field(metadata={"description": "The session"})
    did_key: str = field(metadata={"description": "The did key, i.e. `did:key:z...`"})
    private_key: str = field(
        metadata={"description": "private key corresponding to did_key"}
    )

    async def get(self, url: str, headers: dict = {}):
        """GET for resource

        :param url: url to query
        :param headers: Additional headers"""
        host, target = host_target_from_url(url)

        accept = "application/activity+json"
        date_header = get_gmt_now()

        signature_helper = build_signature(host, "get", target).with_field(
            "date", date_header
        )
        signature_header = signature_helper.ed25519_sign(self.private_key)

        headers = {
            "accept": accept,
            "user-agent": BOVINE_CLIENT_NAME,
            **headers,
            **signature_helper.headers,
            "authorization": f"Moo-Auth-1 {self.did_key}",
            "x-moo-signature": signature_header,
        }

        return await self.session.get(url, headers=headers)

    async def post(
        self,
        url: str,
        body: str,
        headers: dict = {},
        content_type: str = "application/activity+json",
    ):
        """POST to resource

        :param url: The target url
        :param body: The request body
        :param headers: additional request headers.
        :param content_type: The content_type of the body
        """
        host, target = host_target_from_url(url)
        accept = "application/activity+json"
        date_header = get_gmt_now()

        digest = content_digest_sha256(body)

        signature_helper = (
            build_signature(host, "post", target)
            .with_field("date", date_header)
            .with_field("digest", digest)
        )
        signature_header = signature_helper.ed25519_sign(self.private_key)

        headers = {
            "accept": accept,
            "content-type": content_type,
            "user-agent": BOVINE_CLIENT_NAME,
            **headers,
            **signature_helper.headers,
            "authorization": f"Moo-Auth-1 {self.did_key}",
            "x-moo-signature": signature_header,
        }

        return await self.session.post(url, data=body, headers=headers)

    def event_source(self, url: str, headers: dict = {}):
        """Returns an event source

        :param url: url of event source
        :param headers: additional headers"""
        host, target = host_target_from_url(url)
        date_header = get_gmt_now()
        accept = "text/event-stream"
        signature_helper = build_signature(host, "get", target).with_field(
            "date", date_header
        )
        signature_header = signature_helper.ed25519_sign(self.private_key)

        headers = {
            "accept": accept,
            "user-agent": BOVINE_CLIENT_NAME,
            **headers,
            **signature_helper.headers,
            "authorization": f"Moo-Auth-1 {self.did_key}",
            "x-moo-signature": signature_header,
        }

        return EventSource(self.session, url, headers=headers)

event_source 

event_source(url: str, headers: dict = {})

Returns an event source

Parameters:

Name Type Description Default
url str

url of event source

 required
headers dict

additional headers

 {}
Source code in bovine/bovine/clients/moo_auth.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
def event_source(self, url: str, headers: dict = {}):
    """Returns an event source

    :param url: url of event source
    :param headers: additional headers"""
    host, target = host_target_from_url(url)
    date_header = get_gmt_now()
    accept = "text/event-stream"
    signature_helper = build_signature(host, "get", target).with_field(
        "date", date_header
    )
    signature_header = signature_helper.ed25519_sign(self.private_key)

    headers = {
        "accept": accept,
        "user-agent": BOVINE_CLIENT_NAME,
        **headers,
        **signature_helper.headers,
        "authorization": f"Moo-Auth-1 {self.did_key}",
        "x-moo-signature": signature_header,
    }

    return EventSource(self.session, url, headers=headers)

get async 

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

GET for resource

Parameters:

Name Type Description Default
url str

url to query

 required
headers dict

Additional headers

 {}
Source code in bovine/bovine/clients/moo_auth.py 
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
async def get(self, url: str, headers: dict = {}):
    """GET for resource

    :param url: url to query
    :param headers: Additional headers"""
    host, target = host_target_from_url(url)

    accept = "application/activity+json"
    date_header = get_gmt_now()

    signature_helper = build_signature(host, "get", target).with_field(
        "date", date_header
    )
    signature_header = signature_helper.ed25519_sign(self.private_key)

    headers = {
        "accept": accept,
        "user-agent": BOVINE_CLIENT_NAME,
        **headers,
        **signature_helper.headers,
        "authorization": f"Moo-Auth-1 {self.did_key}",
        "x-moo-signature": signature_header,
    }

    return await self.session.get(url, headers=headers)

post async 

post(
    url: str,
    body: str,
    headers: dict = {},
    content_type: str = "application/activity+json",
)

POST to resource

Parameters:

Name Type Description Default
url str

The target url

 required
body str

The request body

 required
headers dict

additional request headers.

 {}
content_type str

The content_type of the body

 'application/activity+json'
Source code in bovine/bovine/clients/moo_auth.py 
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
async def post(
    self,
    url: str,
    body: str,
    headers: dict = {},
    content_type: str = "application/activity+json",
):
    """POST to resource

    :param url: The target url
    :param body: The request body
    :param headers: additional request headers.
    :param content_type: The content_type of the body
    """
    host, target = host_target_from_url(url)
    accept = "application/activity+json"
    date_header = get_gmt_now()

    digest = content_digest_sha256(body)

    signature_helper = (
        build_signature(host, "post", target)
        .with_field("date", date_header)
        .with_field("digest", digest)
    )
    signature_header = signature_helper.ed25519_sign(self.private_key)

    headers = {
        "accept": accept,
        "content-type": content_type,
        "user-agent": BOVINE_CLIENT_NAME,
        **headers,
        **signature_helper.headers,
        "authorization": f"Moo-Auth-1 {self.did_key}",
        "x-moo-signature": signature_header,
    }

    return await self.session.post(url, data=body, headers=headers)