Skip to content

.activity_pub

cattle_grid.activity_pub

cattle_grid.activity_pub.actor

actor_to_object(actor)

Transform the actor to an object

Parameters:

Name Type Description Default
actor Actor
required

Returns:

Type Description
dict
Source code in cattle_grid/activity_pub/actor.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
def actor_to_object(actor: Actor) -> dict:
    """Transform the actor to an object

    :params actor:
    :returns:
    """
    result = AsActor(
        id=actor.actor_id,
        outbox=actor.outbox_uri,
        inbox=actor.inbox_uri,
        followers=actor.followers_uri,
        following=actor.following_uri,
        public_key=actor.public_key,
        public_key_name=actor.public_key_name,
        preferred_username=actor.preferred_username,
        type=actor.profile.get("type", "Person"),
        name=actor.profile.get("name"),
        summary=actor.profile.get("summary"),
        url=actor.profile.get("url"),
        icon=actor.profile.get("image"),
    ).build(visibility=Visibility.OWNER)

    return result

bovine_actor_for_actor_id(actor_id) async

Uses the information stored in [Credential][cattle_grid.ap.models.Credential] to construct a bovine actor

Parameters:

Name Type Description Default
actor_id str
required

Returns:

Type Description
BovineActor | None
Source code in cattle_grid/activity_pub/actor.py 
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
async def bovine_actor_for_actor_id(actor_id: str) -> BovineActor | None:
    """Uses the information stored in [Credential][cattle_grid.ap.models.Credential] to construct a bovine actor

    :params actor_id:
    :returns:
    """
    credential = await Credential.get_or_none(actor_id=actor_id)

    if credential is None:
        return None

    return BovineActor(
        public_key_url=credential.identifier,
        actor_id=actor_id,
        secret=credential.secret,
    )

compute_acct_uri(base_url, preferred_username)

Computes the acct uri

>>> compute_acct_uri("http://host.example/somewhere", "alice")
'acct:alice@host.example'
Source code in cattle_grid/activity_pub/actor.py 
29
30
31
32
33
34
35
36
37
38
39
40
41
def compute_acct_uri(base_url: str, preferred_username: str):
    """Computes the acct uri

    ```pycon
    >>> compute_acct_uri("http://host.example/somewhere", "alice")
    'acct:alice@host.example'

    ```

    """
    host = urlparse(base_url).hostname

    return f"acct:{preferred_username}@{host}"

create_actor(base_url, preferred_username=None, identifiers={}, profile={}) async

Creates a new actor in the database

Source code in cattle_grid/activity_pub/actor.py 
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
async def create_actor(
    base_url: str,
    preferred_username: str | None = None,
    identifiers: dict = {},
    profile: dict = {},
):
    """Creates a new actor in the database"""

    public_key, private_key = generate_rsa_public_private_key()
    public_key_name = "legacy-key-1"
    actor_id = new_url(base_url, "actor")

    actor = await Actor.create(
        actor_id=actor_id,
        inbox_uri=new_url(base_url, "inbox"),
        outbox_uri=new_url(base_url, "outbox"),
        following_uri=new_url(base_url, "following"),
        followers_uri=new_url(base_url, "followers"),
        preferred_username=preferred_username,
        public_key_name=public_key_name,
        public_key=public_key,
        profile=profile,
        automatically_accept_followers=False,
    )
    await Credential.create(
        actor_id=actor_id,
        identifier=f"{actor_id}#{public_key_name}",
        secret=private_key,
    )

    if preferred_username:
        if "webfinger" in identifiers:
            raise ValueError("webfinger key set in identifiers")
        identifiers = {
            **identifiers,
            "webfinger": compute_acct_uri(base_url, preferred_username),
        }

    for name, identifier in identifiers.items():
        await PublicIdentifier.create(actor=actor, name=name, identifier=identifier)

    logging.info("Created actor with id '%s'", actor_id)

    return actor

delete_actor(actor) async

Deletes an actor

Parameters:

Name Type Description Default
actor Actor

Actor to be deleted

 required
Source code in cattle_grid/activity_pub/actor.py 
188
189
190
191
192
193
194
195
196
197
198
async def delete_actor(actor: Actor):
    """Deletes an actor

    :param actor: Actor to be deleted
    """

    # await Credential.filter(actor_id=actor.actor_id).delete()
    await PublicIdentifier.filter(actor=actor).delete()

    actor.status = ActorStatus.deleted
    await actor.save()

delete_for_actor_profile(actor)

Creates a delete activity for the Actor

Source code in cattle_grid/activity_pub/actor.py 
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
def delete_for_actor_profile(actor: Actor) -> dict:
    """Creates a delete activity for the Actor"""

    actor_profile = actor_to_object(actor)
    activity_factory, _ = factories_for_actor_object(actor_profile)

    result = (
        activity_factory.delete(
            actor_profile.get("id"), followers=actor_profile["followers"]
        )
        .as_public()
        .build()
    )

    result["cc"].append(actor_profile["following"])

    return result

followers_for_actor(actor) async

Returns the list of accepted followers

Parameters:

Name Type Description Default
actor Actor
required

Returns:

Type Description
List[str]
Source code in cattle_grid/activity_pub/actor.py 
133
134
135
136
137
138
139
140
141
async def followers_for_actor(actor: Actor) -> List[str]:
    """Returns the list of accepted followers

    :param actor:
    :returns:
    """

    await actor.fetch_related("followers")
    return [x.follower for x in actor.followers if x.accepted]

following_for_actor(actor) async

Returns the list of accepted people to follow said actor. This is the following table.

Parameters:

Name Type Description Default
actor Actor
required

Returns:

Type Description
List[str]
Source code in cattle_grid/activity_pub/actor.py 
144
145
146
147
148
149
150
151
152
153
async def following_for_actor(actor: Actor) -> List[str]:
    """Returns the list of accepted people to follow said actor.
    This is the following table.

    :param actor:
    :returns:
    """

    await actor.fetch_related("following")
    return [x.following for x in actor.following if x.accepted]

remove_from_followers_following(actor_id_to_remove) async

Removes actor_id from all occurring followers and following

Source code in cattle_grid/activity_pub/actor.py 
201
202
203
204
205
async def remove_from_followers_following(actor_id_to_remove: str):
    """Removes actor_id from all occurring followers and following"""

    await Follower.filter(follower=actor_id_to_remove).delete()
    await Following.filter(following=actor_id_to_remove).delete()

update_for_actor_profile(actor)

Creates an update for the Actor

Source code in cattle_grid/activity_pub/actor.py 
156
157
158
159
160
161
162
163
164
165
166
def update_for_actor_profile(actor: Actor) -> dict:
    """Creates an update for the Actor"""

    actor_profile = actor_to_object(actor)
    activity_factory, _ = factories_for_actor_object(actor_profile)

    return (
        activity_factory.update(actor_profile, followers=actor_profile["followers"])
        .as_public()
        .build()
    )

cattle_grid.activity_pub.enqueuer

determine_activity_type(activity)

Determines the type of an activity

>>> determine_activity_type({"type": "Follow"})
'Follow'

>>> determine_activity_type({}) is None
True

In the case of multiple types, these are concatenated. This means that they are probably missed by processing, but don’t get ignored.

>>> determine_activity_type({"type": ["Follow", "WhileSkipping"]})
'FollowWhileSkipping'

Parameters:

Name Type Description Default
activity dict
required

Returns:

Type Description
str | None
Source code in cattle_grid/activity_pub/enqueuer.py 
 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
def determine_activity_type(activity: dict) -> str | None:
    """Determines the type of an activity

    ```pycon
    >>> determine_activity_type({"type": "Follow"})
    'Follow'

    ```


    ```pycon
    >>> determine_activity_type({}) is None
    True

    ```

    In the case of multiple types, these are concatenated. This means
    that they are probably missed by processing, but don't get ignored.

    ```pycon
    >>> determine_activity_type({"type": ["Follow", "WhileSkipping"]})
    'FollowWhileSkipping'

    ```

    :params activity:
    :returns:

    """

    activity_type = activity.get("type")
    if activity_type is None:
        return None
    if isinstance(activity_type, list):
        activity_type = "".join(activity_type)

    return activity_type

enqueue_from_inbox(broker, exchange, receiving_actor_id, content) async

Enqueues a new message arrived from the inbox

The routing key will be incoming.${activity_type}

Source code in cattle_grid/activity_pub/enqueuer.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
async def enqueue_from_inbox(
    broker: RabbitBroker,
    exchange: RabbitExchange,
    receiving_actor_id: str,
    content: dict,
):
    """Enqueues a new message arrived from the inbox

    The routing key will be `incoming.${activity_type}`
    """
    activity_type = determine_activity_type(content)
    if activity_type is None:
        return

    msg = ActivityMessage(
        actor=receiving_actor_id, data=content, activity_type=activity_type
    )

    await broker.publish(
        msg, exchange=exchange, routing_key=f"incoming.{activity_type}"
    )

cattle_grid.activity_pub.models

Data model used to describe ActivityPub related objects

Actor

Bases: Model

Actors administrated by cattle_grid

Source code in cattle_grid/activity_pub/models.py 
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
class Actor(Model):
    """Actors administrated by cattle_grid"""

    id = fields.IntField(primary_key=True)
    actor_id = fields.CharField(max_length=255, unique=True)
    """The id of the actor"""

    inbox_uri = fields.CharField(max_length=255, unique=True)
    """The uri of the inbox"""
    outbox_uri = fields.CharField(max_length=255, unique=True)
    """The uri of the outbox"""
    following_uri = fields.CharField(max_length=255, unique=True)
    """The uri of the following collection"""
    followers_uri = fields.CharField(max_length=255, unique=True)
    """The uri of the followers collection"""

    preferred_username = fields.CharField(max_length=255, null=True)
    """The preferred username, used as the username part of the
    acct-uri of the actor, i.e. `acct:${preferred_username}@domain`.
    See [RFC 7565 The 'acct' URI Scheme](https://www.rfc-editor.org/rfc/rfc7565.html)."""

    public_key_name = fields.CharField(max_length=255)
    """The name given to the public key, i.e. the id will be
    `${actor_id}#${public_key_name}."""
    public_key = fields.TextField()
    """The public key"""

    automatically_accept_followers = fields.BooleanField()
    """Set to true to indicate cattle_grid should automatically
    accept follow requests"""
    profile: Dict[str, Any] = fields.JSONField()
    """Additional profile values"""

    status = fields.CharEnumField(ActorStatus, default=ActorStatus.active)
    """Represents the status of the actor"""

    followers: fields.ReverseRelation["Follower"]
    following: fields.ReverseRelation["Following"]

actor_id = fields.CharField(max_length=255, unique=True) class-attribute instance-attribute

The id of the actor

automatically_accept_followers = fields.BooleanField() class-attribute instance-attribute

Set to true to indicate cattle_grid should automatically accept follow requests

followers_uri = fields.CharField(max_length=255, unique=True) class-attribute instance-attribute

The uri of the followers collection

following_uri = fields.CharField(max_length=255, unique=True) class-attribute instance-attribute

The uri of the following collection

inbox_uri = fields.CharField(max_length=255, unique=True) class-attribute instance-attribute

The uri of the inbox

outbox_uri = fields.CharField(max_length=255, unique=True) class-attribute instance-attribute

The uri of the outbox

preferred_username = fields.CharField(max_length=255, null=True) class-attribute instance-attribute

The preferred username, used as the username part of the acct-uri of the actor, i.e. acct:${preferred_username}@domain. See RFC 7565 The ‘acct’ URI Scheme.

profile: Dict[str, Any] = fields.JSONField() class-attribute instance-attribute

Additional profile values

public_key = fields.TextField() class-attribute instance-attribute

The public key

public_key_name = fields.CharField(max_length=255) class-attribute instance-attribute

The name given to the public key, i.e. the id will be `${actor_id}#${public_key_name}.

status = fields.CharEnumField(ActorStatus, default=ActorStatus.active) class-attribute instance-attribute

Represents the status of the actor

ActorStatus

Bases: StrEnum

Represents the status of the actor

Source code in cattle_grid/activity_pub/models.py 
15
16
17
18
19
class ActorStatus(StrEnum):
    """Represents the status of the actor"""

    active = auto()
    deleted = auto()

Credential

Bases: Model

The secrets of the actor

Source code in cattle_grid/activity_pub/models.py 
121
122
123
124
125
126
127
128
129
class Credential(Model):
    """The secrets of the actor"""

    id = fields.IntField(primary_key=True)

    actor_id = fields.CharField(max_length=255)
    identifier = fields.CharField(max_length=255)

    secret = fields.TextField()

Follower

Bases: Model

The people that follow the actor

Source code in cattle_grid/activity_pub/models.py 
77
78
79
80
81
82
83
84
85
86
87
88
class Follower(Model):
    """The people that follow the actor"""

    id = fields.IntField(primary_key=True)

    actor: fields.ForeignKeyRelation[Actor] = fields.ForeignKeyField(
        "ap_models.Actor", related_name="followers"
    )

    follower = fields.CharField(max_length=255)
    request = fields.CharField(max_length=255)
    accepted = fields.BooleanField()

Following

Bases: Model

The people the actor is following

Source code in cattle_grid/activity_pub/models.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
class Following(Model):
    """The people the actor is following"""

    id = fields.IntField(primary_key=True)

    actor: fields.ForeignKeyRelation[Actor] = fields.ForeignKeyField(
        "ap_models.Actor", related_name="following"
    )

    following = fields.CharField(max_length=255)
    request = fields.CharField(max_length=255)
    accepted = fields.BooleanField()

InboxLocation

Bases: Model

Describes the location of an inbox. Used to send ActivityPub Activities addressed to the actor to the corresponding inbox.

This information is also collected for remote actors.

Source code in cattle_grid/activity_pub/models.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
class InboxLocation(Model):
    """Describes the location of an inbox. Used to send
    ActivityPub Activities addressed to the actor to the
    corresponding inbox.

    This information is also collected for remote actors.
    """

    id = fields.IntField(primary_key=True)
    actor = fields.CharField(max_length=255, unique=True)
    """The id of the remote actor"""
    inbox = fields.CharField(max_length=255)
    """The inbox of the remote actor"""

actor = fields.CharField(max_length=255, unique=True) class-attribute instance-attribute

The id of the remote actor

inbox = fields.CharField(max_length=255) class-attribute instance-attribute

The inbox of the remote actor

PublicIdentifier

Bases: Model

Public identifiers

Source code in cattle_grid/activity_pub/models.py 
105
106
107
108
109
110
111
112
113
114
115
116
117
118
class PublicIdentifier(Model):
    """Public identifiers"""

    id = fields.IntField(primary_key=True)

    actor: fields.ForeignKeyRelation[Actor] = fields.ForeignKeyField(
        "ap_models.Actor", related_name="identifiers"
    )
    """The actor the public key belongs to"""

    name = fields.CharField(max_length=255)
    """name of public identifier"""
    identifier = fields.CharField(max_length=255, unique=True)
    """The public identifier, e.g. an acct-uri"""

actor: fields.ForeignKeyRelation[Actor] = fields.ForeignKeyField('ap_models.Actor', related_name='identifiers') class-attribute instance-attribute

The actor the public key belongs to

identifier = fields.CharField(max_length=255, unique=True) class-attribute instance-attribute

The public identifier, e.g. an acct-uri

name = fields.CharField(max_length=255) class-attribute instance-attribute

name of public identifier

StoredActivity

Bases: Model

cattle_grid generates activities under some circumstances (see FIXME). These will be stored in this table

Source code in cattle_grid/activity_pub/models.py 
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
class StoredActivity(Model):
    """cattle_grid generates activities under some
    circumstances (see FIXME). These will be stored
    in this table"""

    id = fields.CharField(max_length=255, primary_key=True)

    actor: fields.ForeignKeyRelation[Actor] = fields.ForeignKeyField("ap_models.Actor")
    """The actor this activity orginates from"""

    data = fields.JSONField()
    """The activity"""

    published = fields.DatetimeField()
    """When the activity was published"""

actor: fields.ForeignKeyRelation[Actor] = fields.ForeignKeyField('ap_models.Actor') class-attribute instance-attribute

The actor this activity orginates from

data = fields.JSONField() class-attribute instance-attribute

The activity

published = fields.DatetimeField() class-attribute instance-attribute

When the activity was published