Skip to content

.activity_pub.processing

cattle_grid.activity_pub.processing

incoming

incoming_accept_activity(msg, actor, broker=Context()) async

Handles an incoming Accept activity

Source code in cattle_grid/activity_pub/processing/incoming.py 
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
async def incoming_accept_activity(
    msg: ActivityMessage, actor: MessageActor, broker=Context()
):
    """Handles an incoming Accept activity"""
    accept_request = msg.data
    request_being_accepted = id_for_object(accept_request.get("object"))

    following = await Following.get_or_none(request=request_being_accepted)

    if not following:
        logger.warning("Follow request with id '%s' not found", request_being_accepted)
        return

    following.accepted = True
    await following.save()

    logger.info("Processed follow request %s (following)", request_being_accepted)

incoming_block_activity(msg, actor, broker=Context()) async

Handles an incoming Block activity

Source code in cattle_grid/activity_pub/processing/incoming.py 
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
async def incoming_block_activity(
    msg: ActivityMessage, actor: MessageActor, broker=Context()
):
    """Handles an incoming Block activity"""
    current_actor_id = msg.data.get("object")
    if current_actor_id != actor.actor_id:
        logger.warning("Mismatch of actor and target of block")
        return

    actor_blocking = msg.data.get("actor")

    following = await Following.get_or_none(following=actor_blocking)

    if not following:
        return

    await following.delete()

incoming_follow_request(msg, actor, broker=Context()) async

For an incoming Follow request an entry in the Follower table is created with having accepted set to False.

If the actor automatically accepts followers, the actor sends Accept activity to the actor requesting to follow it.

Source code in cattle_grid/activity_pub/processing/incoming.py 
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
async def incoming_follow_request(
    msg: ActivityMessage, actor: MessageActor, broker=Context()
):
    """
    For an incoming Follow request an entry in the Follower table is created
    with having `accepted` set to False.

    If the actor automatically accepts followers, the actor sends Accept activity
    to the actor requesting to follow it.
    """
    follow_request = msg.data
    to_follow = follow_request.get("object")
    if isinstance(to_follow, dict):
        to_follow = to_follow.get("id")

    if to_follow is None or to_follow != actor.actor_id:
        return

    request_id = follow_request.get("id")
    follower = follow_request.get("actor")

    await Follower.update_or_create(
        actor=actor,
        follower=follower,
        defaults={"request": request_id, "accepted": False},
    )

    if actor.automatically_accept_followers:
        actor_object = actor_to_object(actor)
        activity_factory, _ = factories_for_actor_object(actor_object)
        accept = activity_factory.accept(follow_request).build()

        await broker.publish(
            StoreActivityMessage(actor=actor.actor_id, data=accept),
            routing_key="store_activity",
            exchange=internal_exchange(),
        )
        logger.info(
            "Got follow request from %s with id %s (auto accepted)",
            follower,
            request_id,
        )

    else:
        logger.info("Got follow request from %s with id %s", follower, request_id)

incoming_reject_activity(msg, actor, broker=Context()) async

Handles an incoming Reject activity

Source code in cattle_grid/activity_pub/processing/incoming.py 
105
106
107
108
109
110
111
112
113
114
115
116
117
async def incoming_reject_activity(
    msg: ActivityMessage, actor: MessageActor, broker=Context()
):
    """Handles an incoming Reject activity"""
    reject_request = msg.data
    request_being_rejected = reject_request.get("object")

    following = await Following.get_or_none(request=request_being_rejected)

    if not following:
        return

    await following.delete()

incoming_undo_activity(msg, actor, broker=Context()) async

Handles an incoming Undo activity

Source code in cattle_grid/activity_pub/processing/incoming.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
async def incoming_undo_activity(
    msg: ActivityMessage, actor: MessageActor, broker=Context()
):
    """Handles an incoming Undo activity"""
    accept_request = msg.data
    request_being_undone = accept_request.get("object")

    follower = await Follower.get_or_none(request=request_being_undone)

    if not follower:
        return

    await follower.delete()

outgoing

outgoing_accept_request(msg, actor, broker=Context()) async

Handles an outgoing Accept activity

Source code in cattle_grid/activity_pub/processing/outgoing.py 
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
async def outgoing_accept_request(
    msg: ActivityMessage, actor: MessageActor, broker=Context()
):
    """Handles an outgoing Accept activity"""
    accept_request = msg.data
    request_being_accepted = id_for_object(accept_request.get("object"))

    follower = await Follower.get_or_none(request=request_being_accepted)

    if not follower:
        logger.warning("Follow request with id '%s' not found", request_being_accepted)
        return

    follower.accepted = True
    await follower.save()

    logger.info("Accepted follow request %s", request_being_accepted)

outgoing_block_activity(msg, actor, broker=Context()) async

Handles an outgoing Block activity

Source code in cattle_grid/activity_pub/processing/outgoing.py 
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
async def outgoing_block_activity(
    msg: ActivityMessage, actor: MessageActor, broker=Context()
):
    """Handles an outgoing Block activity"""
    block_request = msg.data
    actor_being_blocked = block_request.get("object")

    follower = await Follower.get_or_none(follower=actor_being_blocked)

    if follower:
        await follower.delete()

    block_id = block_request.get("id", "permanent")

    if block_id == "permanent":
        logger.warning("%s permanently blocked %s", actor.actor_id, actor_being_blocked)

    await Blocking.create(
        actor=actor, blocking=actor_being_blocked, request=block_id, active=True
    )
    logger.info("Created block")
    logger.info("%s blocked %s", actor.actor_id, actor_being_blocked)

outgoing_follow_request(msg, actor, broker=Context()) async

Handles an outgoing Follow request

Source code in cattle_grid/activity_pub/processing/outgoing.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
async def outgoing_follow_request(
    msg: ActivityMessage, actor: MessageActor, broker=Context()
):
    """Handles an outgoing Follow request"""

    follow_request = msg.data
    to_follow = follow_request.get("object")
    if isinstance(to_follow, dict):
        to_follow = to_follow.get("id")

    if to_follow is None:
        return

    logger.info("Send follow request to %s", to_follow)

    await Following.update_or_create(
        actor=actor,
        following=to_follow,
        defaults={"request": follow_request.get("id"), "accepted": False},
    )

outgoing_message_distribution(msg, broker=Context()) async

Distributes the message to its recipients

Source code in cattle_grid/activity_pub/processing/outgoing.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
async def outgoing_message_distribution(msg: ActivityMessage, broker=Context()):
    """Distributes the message to its recipients"""

    recipients = recipients_for_object(msg.data)
    recipients = remove_public(recipients)

    logger.debug("Got recipients %s", ", ".join(recipients))

    recipients = await update_recipients_for_collections(msg, recipients)

    exchange = internal_exchange()
    logger.info("Got exchange")
    logger.info(exchange)

    for recipient in recipients:
        await broker.publish(
            ToSendMessage(actor=msg.actor, data=msg.data, target=recipient),
            exchange=exchange,
            routing_key="to_send",
        )

outgoing_reject_activity(msg, actor, broker=Context()) async

Handles an outgoing Reject activity

Source code in cattle_grid/activity_pub/processing/outgoing.py 
106
107
108
109
110
111
112
113
114
115
116
async def outgoing_reject_activity(
    msg: ActivityMessage, actor: MessageActor, broker=Context()
):
    """Handles an outgoing Reject activity"""
    reject_request = msg.data
    request_being_rejected = reject_request.get("object")

    follower = await Follower.get_or_none(request=request_being_rejected)
    if follower:
        await follower.delete()
        return

outgoing_undo_request(msg, actor, broker=Context()) async

Handles an outgoing Undo activity

Source code in cattle_grid/activity_pub/processing/outgoing.py 
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
async def outgoing_undo_request(
    msg: ActivityMessage, actor: MessageActor, broker=Context()
):
    """Handles an outgoing Undo activity"""
    accept_request = msg.data
    request_being_undone = accept_request.get("object")

    following = await Following.get_or_none(request=request_being_undone)
    if following:
        await following.delete()
        return

    blocking = await Blocking.get_or_none(request=request_being_undone)
    if blocking:
        blocking.active = False
        await blocking.save()
        return

remote

fetch_object(msg, actor, lookup) async

Handles retrieving a remote object

Source code in cattle_grid/activity_pub/processing/remote.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
async def fetch_object(
    msg: FetchMessage,
    actor: MessageBovineActor,
    lookup: LookupAnnotation,
):
    """Handles retrieving a remote object"""

    try:
        lookup_result = await lookup(Lookup(uri=msg.uri, actor=msg.actor))
        if lookup_result.result:
            return lookup_result.result
    except Exception as e:
        logger.error("Something went up with lookup")
        logger.exception(e)

    result = await actor.get(lookup_result.uri, fail_silently=True)

    return result

resolve_inbox(actor, target) async

Resolves the inbox of target for actor using a cache

Source code in cattle_grid/activity_pub/processing/remote.py 
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
async def resolve_inbox(actor, target):
    """Resolves the inbox of target for actor using
    a cache"""
    cached = await InboxLocation.get_or_none(actor=target)
    if cached:
        return cached.inbox

    target_actor = await actor.get(target)
    if not target_actor:
        return None

    inbox = target_actor.get("inbox")
    if inbox is None:
        return

    await InboxLocation.update_or_create(actor=target, defaults={"inbox": inbox})

    return inbox

sending_message(msg, actor) async

Handles sending a message

Source code in cattle_grid/activity_pub/processing/remote.py 
56
57
58
59
60
61
62
63
64
async def sending_message(
    msg: ToSendMessage,
    actor: MessageBovineActor,
):
    """Handles sending a message"""
    inbox = await resolve_inbox(actor, msg.target)
    if inbox:
        result = await actor.post(inbox, msg.data)
        logger.info("Got %s for sending to %s", str(result), inbox)

store_activity

store_activity_subscriber(msg, actor, broker=Context()) async

This method used internally to store activities generated by cattle_grid, e.g. an automatically generated accept requests. We note that it assigns an id to the activity.

After storing the activity, it is published to the appropriate outgoing.* topic, which triggers sending the activity.

Source code in cattle_grid/activity_pub/processing/store_activity.py 
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
async def store_activity_subscriber(
    msg: StoreActivityMessage,
    actor: MessageActor,
    broker=Context(),
):
    """This method used internally to store activities
    generated by `cattle_grid`, e.g. an automatically generated
    accept requests. We note that it assigns an id to
    the activity.

    After storing the activity, it is published to the
    appropriate `outgoing.*` topic, which triggers sending
    the activity.
    """
    base_url = "/".join(actor.actor_id.split("/")[:-2])
    uuid = str(uuid7())
    activity_id = f"{base_url}/object/{uuid}"

    activity = msg.data
    activity["id"] = activity_id
    activity_type = activity.get("type")

    await StoredActivity.create(
        id=uuid, actor=actor, data=activity, published=datetime.now()
    )

    await broker.publish(
        {"actor": actor.actor_id, "data": activity},
        routing_key=f"outgoing.{activity_type}",
        exchange=internal_exchange(),
    )

util

update_recipients_for_collections(msg, recipients) async

Updates recipients with followers and following collection.

Source code in cattle_grid/activity_pub/processing/util.py 
18
19
20
21
22
23
24
25
26
27
28
async def update_recipients_for_collections(
    msg: ActivityMessage, recipients: Set[str]
) -> Set[str]:
    """Updates recipients with followers and following collection."""

    db_actor = await Actor.get(actor_id=msg.actor)
    self_delete = actor_deletes_themselves(msg.data)

    return await update_recipients_for_actor(
        db_actor, recipients, self_delete=self_delete
    )