Skip to content

bovine_store

BovineAdminStore

Store for managing actors. This store should be used to create actor management interfaces. Usage:

store = BovineAdmineStore(domain="cows.example")
bovine_name = await store.register("moocow")
await store.add_identity_string_to_actor(
    bovine_name, "sample", "did:example:123")

This will create the account @moocow@cows.example which can be accessed through Moo-Auth-1 with the secret corresponding to the did.

These can be kept separate from the actual fediverse server implementation.

For usage see bovine_tool and bovine_management.

The parameters endpoint_path, endpoint_path_function can probably be removed at one point. The parameter domain can actually be of the form “http://domain_name” or “https://domain_name”. This is useful for end to end tests.

Source code in bovine_store/bovine_store/__init__.py 
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
class BovineAdminStore:
    """Store for managing actors. This store should be used to create
    actor management interfaces. Usage:

    ```python
    store = BovineAdmineStore(domain="cows.example")
    bovine_name = await store.register("moocow")
    await store.add_identity_string_to_actor(
        bovine_name, "sample", "did:example:123")
    ```

    This will create the account `@moocow@cows.example` which can be accessed
    through Moo-Auth-1 with the secret corresponding to the did.

    These can be kept separate from the actual fediverse server implementation.

    For usage see `bovine_tool` and `bovine_management`.

    The parameters endpoint_path, endpoint_path_function can probably be removed
    at one point. The parameter domain can actually be of the form
    "http://domain_name" or "https://domain_name". This is useful for end
    to end tests.
    """

    def __init__(
        self,
        endpoint_path: str | None = None,
        endpoint_path_function: Callable | None = None,
        domain: str | None = None,
    ):
        self.domain = domain
        self.endpoint_path_function = endpoint_path_function
        if endpoint_path_function is None and endpoint_path:
            self.endpoint_path_function = endpoint_path_function_from_endpoint(
                endpoint_path
            )

    async def register(self, handle_name: str, endpoints: dict = {}) -> str:
        """registers a new user with handle_name. The domain is extracted from
        the endpoint path configured when creating the store. This method
        generates the necessary private keys for the actor. Implementations
        of bovine should never require one adding a secret.

        :param handle_name: The username
        :param endpoints: dictionary of predefined endpoints.

        :return: The bovine name given by handle_name + uuid4"""

        if self.endpoint_path_function is None:
            if self.domain is None:
                raise Exception("Need to specify an endpoint")
            if "://" not in self.domain:
                domain = "https://" + self.domain
            else:
                domain = self.domain

            self.endpoint_path_function = endpoint_path_function_from_endpoint(
                urljoin(domain, "endpoints/template")
            )

        result = await register(
            handle_name,
            endpoint_path_function=self.endpoint_path_function,
            endpoints=endpoints,
        )
        return result.bovine_name

    async def add_identity_string_to_actor(
        self, bovine_name: str, name: str, identity: str
    ) -> None:
        """Modifies an Actor by adding a new identity to it. name is used
        to identity the identity and serves little functional purpose."""
        actor = await BovineActor.get_or_none(bovine_name=bovine_name)
        await BovineActorKeyPair.create(
            bovine_actor=actor, name=name, public_key=identity, private_key=""
        )

    async def set_properties_for_actor(
        self, bovine_name: str, properties: dict
    ) -> None:
        """Sets a new properties object for the actor"""
        actor = await BovineActor.get_or_none(bovine_name=bovine_name)
        if actor is None:
            return None
        actor.properties = properties
        await actor.save()

    async def actor_for_name(self, bovine_name: str) -> Optional[BovineStoreActor]:
        """Retrieves the actor for a given bovine_name"""

        actor = await BovineActor.get_or_none(bovine_name=bovine_name).prefetch_related(
            "endpoints", "keypairs"
        )
        if actor is None:
            return None

        return await BovineStoreActor.from_database(actor, None)

    async def list_bovine_names(self) -> List[str]:
        result = await BovineActor.filter().all()
        return [x.bovine_name for x in result]

actor_for_name(bovine_name) async

Retrieves the actor for a given bovine_name

Source code in bovine_store/bovine_store/__init__.py 
213
214
215
216
217
218
219
220
221
222
async def actor_for_name(self, bovine_name: str) -> Optional[BovineStoreActor]:
    """Retrieves the actor for a given bovine_name"""

    actor = await BovineActor.get_or_none(bovine_name=bovine_name).prefetch_related(
        "endpoints", "keypairs"
    )
    if actor is None:
        return None

    return await BovineStoreActor.from_database(actor, None)

add_identity_string_to_actor(bovine_name, name, identity) async

Modifies an Actor by adding a new identity to it. name is used to identity the identity and serves little functional purpose.

Source code in bovine_store/bovine_store/__init__.py 
193
194
195
196
197
198
199
200
201
async def add_identity_string_to_actor(
    self, bovine_name: str, name: str, identity: str
) -> None:
    """Modifies an Actor by adding a new identity to it. name is used
    to identity the identity and serves little functional purpose."""
    actor = await BovineActor.get_or_none(bovine_name=bovine_name)
    await BovineActorKeyPair.create(
        bovine_actor=actor, name=name, public_key=identity, private_key=""
    )

register(handle_name, endpoints={}) async

registers a new user with handle_name. The domain is extracted from the endpoint path configured when creating the store. This method generates the necessary private keys for the actor. Implementations of bovine should never require one adding a secret.

Parameters:

Name Type Description Default
handle_name str

The username

 required
endpoints dict

dictionary of predefined endpoints.

 {}

Returns:

Type Description
str

The bovine name given by handle_name + uuid4
Source code in bovine_store/bovine_store/__init__.py 
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
async def register(self, handle_name: str, endpoints: dict = {}) -> str:
    """registers a new user with handle_name. The domain is extracted from
    the endpoint path configured when creating the store. This method
    generates the necessary private keys for the actor. Implementations
    of bovine should never require one adding a secret.

    :param handle_name: The username
    :param endpoints: dictionary of predefined endpoints.

    :return: The bovine name given by handle_name + uuid4"""

    if self.endpoint_path_function is None:
        if self.domain is None:
            raise Exception("Need to specify an endpoint")
        if "://" not in self.domain:
            domain = "https://" + self.domain
        else:
            domain = self.domain

        self.endpoint_path_function = endpoint_path_function_from_endpoint(
            urljoin(domain, "endpoints/template")
        )

    result = await register(
        handle_name,
        endpoint_path_function=self.endpoint_path_function,
        endpoints=endpoints,
    )
    return result.bovine_name

set_properties_for_actor(bovine_name, properties) async

Sets a new properties object for the actor

Source code in bovine_store/bovine_store/__init__.py 
203
204
205
206
207
208
209
210
211
async def set_properties_for_actor(
    self, bovine_name: str, properties: dict
) -> None:
    """Sets a new properties object for the actor"""
    actor = await BovineActor.get_or_none(bovine_name=bovine_name)
    if actor is None:
        return None
    actor.properties = properties
    await actor.save()

BovineStore

Basic interface for the bovine store, should be used to obtain actors.

Source code in bovine_store/bovine_store/__init__.py 
 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
115
116
117
118
119
120
class BovineStore:
    """Basic interface for the bovine store, should be used to obtain actors."""

    def __init__(
        self,
        host: str | None = None,
        session: aiohttp.ClientSession | None = None,
        application_data: dict = {},
    ):
        self.session = session
        self.application_data = application_data
        self.domain = domain_from_host(host)

    async def retrieve(self, retriever: str, object_id: str, include=[]):
        """Retrieves the object identified by object_it for the retriever. This is used
        when a GET request happens, as it is unclear to which actor the
        object_id belongs."""
        return await retrieve_remote_object(retriever, object_id, include=include)

    async def retrieve_for_get(self, retriever, object_id):
        """Retrieves an object"""
        return await retrieve_remote_object(retriever, object_id)

    async def actor_for_name(self, bovine_name: str) -> Optional[BovineStoreActor]:
        """Retrieves the actor for a given bovine_name"""

        actor = await BovineActor.get_or_none(bovine_name=bovine_name).prefetch_related(
            "endpoints", "keypairs"
        )
        if actor is None:
            return None

        return await BovineStoreActor.from_database(actor, self.session)

    @lru_cache
    async def resolve_endpoint(
        self, endpoint: str
    ) -> Tuple[Optional[EndpointType], Optional[BovineStoreActor]]:
        """Given the url of an endpoint, returns the type of the endpoint, e.g. inbox,
        and the corresponding BovineStoreActor"""
        actor_endpoint = await BovineActorEndpoint.get_or_none(
            name=endpoint
        ).prefetch_related("bovine_actor")

        if not actor_endpoint:
            return None, None
        await actor_endpoint.bovine_actor.fetch_related("endpoints", "keypairs")
        return actor_endpoint.endpoint_type, await BovineStoreActor.from_database(
            actor_endpoint.bovine_actor, self.session
        )

    async def resolve_endpoint_no_cache(
        self, endpoint: str
    ) -> Tuple[Optional[EndpointType], Optional[BovineStoreActor]]:
        """Given the url of an endpoint, returns the type of the endpoint, e.g. inbox,
        and the corresponding BovineStoreActor"""
        actor_endpoint = await BovineActorEndpoint.get_or_none(
            name=endpoint
        ).prefetch_related("bovine_actor")

        if not actor_endpoint:
            return None, None
        await actor_endpoint.bovine_actor.fetch_related("endpoints", "keypairs")
        return actor_endpoint.endpoint_type, await BovineStoreActor.from_database(
            actor_endpoint.bovine_actor, self.session
        )

    async def user_count(self) -> int:
        """Returns the total number of bovine actors used in nodeinfo,
        this count probably lies with bovine being a multi hostname server
        and supporting bot accounts."""
        return await BovineActor.filter().count()

    async def get_account_url_for_identity(self, identity: str):
        """Returns the account, i.e. the identity starting with acct:,
        and the actoru_url for a given identity"""
        key = await BovineActorKeyPair.get_or_none(
            public_key=identity
        ).prefetch_related("bovine_actor")
        if not key:
            return None, None
        await key.bovine_actor.fetch_related("endpoints", "keypairs")
        return user_to_account_and_actor_url(key.bovine_actor)

    async def application_actor_for_url(self, url: str):
        """Returns the application actor fot the given url"""
        account_url = urljoin(url, "/activitypub/bovine")
        application_data = await load_application_actor_data(self.application_data)

        app_actor = bovine_store.actor.BovineApplicationActor(
            {**application_data, "account_url": account_url}
        )
        await app_actor.init(session=self.session)

        return app_actor

actor_for_name(bovine_name) async

Retrieves the actor for a given bovine_name

Source code in bovine_store/bovine_store/__init__.py 
49
50
51
52
53
54
55
56
57
58
async def actor_for_name(self, bovine_name: str) -> Optional[BovineStoreActor]:
    """Retrieves the actor for a given bovine_name"""

    actor = await BovineActor.get_or_none(bovine_name=bovine_name).prefetch_related(
        "endpoints", "keypairs"
    )
    if actor is None:
        return None

    return await BovineStoreActor.from_database(actor, self.session)

application_actor_for_url(url) async

Returns the application actor fot the given url

Source code in bovine_store/bovine_store/__init__.py 
110
111
112
113
114
115
116
117
118
119
120
async def application_actor_for_url(self, url: str):
    """Returns the application actor fot the given url"""
    account_url = urljoin(url, "/activitypub/bovine")
    application_data = await load_application_actor_data(self.application_data)

    app_actor = bovine_store.actor.BovineApplicationActor(
        {**application_data, "account_url": account_url}
    )
    await app_actor.init(session=self.session)

    return app_actor

get_account_url_for_identity(identity) async

Returns the account, i.e. the identity starting with acct:, and the actoru_url for a given identity

Source code in bovine_store/bovine_store/__init__.py 
 99
100
101
102
103
104
105
106
107
108
async def get_account_url_for_identity(self, identity: str):
    """Returns the account, i.e. the identity starting with acct:,
    and the actoru_url for a given identity"""
    key = await BovineActorKeyPair.get_or_none(
        public_key=identity
    ).prefetch_related("bovine_actor")
    if not key:
        return None, None
    await key.bovine_actor.fetch_related("endpoints", "keypairs")
    return user_to_account_and_actor_url(key.bovine_actor)

resolve_endpoint(endpoint) async

Given the url of an endpoint, returns the type of the endpoint, e.g. inbox, and the corresponding BovineStoreActor

Source code in bovine_store/bovine_store/__init__.py 
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
@lru_cache
async def resolve_endpoint(
    self, endpoint: str
) -> Tuple[Optional[EndpointType], Optional[BovineStoreActor]]:
    """Given the url of an endpoint, returns the type of the endpoint, e.g. inbox,
    and the corresponding BovineStoreActor"""
    actor_endpoint = await BovineActorEndpoint.get_or_none(
        name=endpoint
    ).prefetch_related("bovine_actor")

    if not actor_endpoint:
        return None, None
    await actor_endpoint.bovine_actor.fetch_related("endpoints", "keypairs")
    return actor_endpoint.endpoint_type, await BovineStoreActor.from_database(
        actor_endpoint.bovine_actor, self.session
    )

resolve_endpoint_no_cache(endpoint) async

Given the url of an endpoint, returns the type of the endpoint, e.g. inbox, and the corresponding BovineStoreActor

Source code in bovine_store/bovine_store/__init__.py 
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
async def resolve_endpoint_no_cache(
    self, endpoint: str
) -> Tuple[Optional[EndpointType], Optional[BovineStoreActor]]:
    """Given the url of an endpoint, returns the type of the endpoint, e.g. inbox,
    and the corresponding BovineStoreActor"""
    actor_endpoint = await BovineActorEndpoint.get_or_none(
        name=endpoint
    ).prefetch_related("bovine_actor")

    if not actor_endpoint:
        return None, None
    await actor_endpoint.bovine_actor.fetch_related("endpoints", "keypairs")
    return actor_endpoint.endpoint_type, await BovineStoreActor.from_database(
        actor_endpoint.bovine_actor, self.session
    )

retrieve(retriever, object_id, include=[]) async

Retrieves the object identified by object_it for the retriever. This is used when a GET request happens, as it is unclear to which actor the object_id belongs.

Source code in bovine_store/bovine_store/__init__.py 
39
40
41
42
43
async def retrieve(self, retriever: str, object_id: str, include=[]):
    """Retrieves the object identified by object_it for the retriever. This is used
    when a GET request happens, as it is unclear to which actor the
    object_id belongs."""
    return await retrieve_remote_object(retriever, object_id, include=include)

retrieve_for_get(retriever, object_id) async

Retrieves an object

Source code in bovine_store/bovine_store/__init__.py 
45
46
47
async def retrieve_for_get(self, retriever, object_id):
    """Retrieves an object"""
    return await retrieve_remote_object(retriever, object_id)

user_count() async

Returns the total number of bovine actors used in nodeinfo, this count probably lies with bovine being a multi hostname server and supporting bot accounts.

Source code in bovine_store/bovine_store/__init__.py 
93
94
95
96
97
async def user_count(self) -> int:
    """Returns the total number of bovine actors used in nodeinfo,
    this count probably lies with bovine being a multi hostname server
    and supporting bot accounts."""
    return await BovineActor.filter().count()

bovine_admin_store(db_url, domain=None) async

Allows one to manage a BovineAdminStore using an async context manager.

Source code in bovine_store/bovine_store/__init__.py 
229
230
231
232
233
234
235
236
237
238
@asynccontextmanager
async def bovine_admin_store(db_url: str, domain: str = None):
    """Allows one to manage a [`BovineAdminStore`][bovine_store.BovineAdminStore]
    using an async context manager."""

    await Tortoise.init(config=tortoise_config(db_url))

    yield BovineAdminStore(domain=domain)

    await Tortoise.close_connections()