Skip to content

.activity_pub.muck_out

cattle_grid.activity_pub.muck_out

This package contains tools to turn ActivityPub messages into something cleaner. Its takes include

  • Normalization
  • Validation

cattle_grid.activity_pub.muck_out.normalize

Routings to normalize an ActivityPub activity

normalize_activity(activity, actor=None)

Normalizes activities.

Parameters:

Name Type Description Default
activity dict

The activity being normalized

 required
actor str | None

Actor receiving this activity

 None

Returns:

Type Description
Activity
Source code in cattle_grid/activity_pub/muck_out/normalize.py 
 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
def normalize_activity(activity: dict, actor: str | None = None) -> Activity:
    """
    Normalizes activities.

    :param activity: The activity being normalized
    :param actor: Actor receiving this activity
    :returns:
    """
    try:
        obj = activity.get("object")
        if isinstance(obj, dict):
            try:
                obj = normalize_object(obj)
            except Exception:
                if isinstance(obj, dict):
                    obj = obj.get("id")

        return Activity.model_validate(
            {
                "@context": activity.get("@context"),
                "id": normalize_id(activity),
                "type": activity.get("type"),
                "actor": id_for_object(activity.get("actor")),
                "object": obj,
                "to": normalize_to(activity.get("to"), actor),
                "cc": list_from_value(activity.get("cc")),
                "published": activity.get("published"),
                "target": activity.get("target"),
                "content": activity.get("content"),
            }
        )
    except Exception as e:
        logger.exception(e)
        logger.info(activity)

        raise e

normalize_id(activity)

Creates a normalized id

>>> normalize_id({"id": "http://id.example"})
'http://id.example'

>>> normalize_id({})
Traceback (most recent call last):
    ...
ValueError: Cannot fake id if actor is not present
Source code in cattle_grid/activity_pub/muck_out/normalize.py 
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
def normalize_id(activity):
    """
    Creates a normalized id

    ```pycon
    >>> normalize_id({"id": "http://id.example"})
    'http://id.example'

    >>> normalize_id({})
    Traceback (most recent call last):
        ...
    ValueError: Cannot fake id if actor is not present

    ```
    """
    result = activity.get("id")
    if result is not None:
        return result
    actor_id = id_for_object(activity.get("actor"))

    if actor_id is None:
        raise ValueError("Cannot fake id if actor is not present")

    return f"{actor_id}#fake_id" + str(uuid7())

normalize_object(obj)

Normalizes an object

Parameters:

Name Type Description Default
obj dict

The object to be normalized

 required

Returns:

Type Description
Object
Source code in cattle_grid/activity_pub/muck_out/normalize.py 
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
def normalize_object(obj: dict) -> Object:
    """Normalizes an object

    :params obj: The object to be normalized
    :returns:
    """
    return Object.model_validate(
        {
            "@context": obj.get("@context"),
            "id": obj.get("id"),
            "type": obj.get("type"),
            "attributedTo": id_for_object(obj.get("attributedTo")),
            "obj": obj.get("obj"),
            "to": list_from_value(obj.get("to")),
            "cc": list_from_value(obj.get("cc")),
            "published": obj.get("published"),
            "target": obj.get("target"),
            "content": sanitize_html(obj.get("content")),
            "attachment": list_from_value(obj.get("attachment")),
            "sensitive": obj.get("sensitive"),
            "summary": sanitize_html(obj.get("summary")),
            "tag": list_from_value(obj.get("tag")),
            "url": normalize_url(obj.get("url")),
            "inReplyTo": id_for_object(obj.get("inReplyTo")),
        }
    )

normalize_to(value, actor)

Normalizes the to value

>>> normalize_to(None, "http://actor.example")
['http://actor.example']

>>> normalize_to("http://to.example", None)
['http://to.example']

>>> normalize_to(["http://alice.example", "http://bob.example"], None)
['http://alice.example', 'http://bob.example']
Source code in cattle_grid/activity_pub/muck_out/normalize.py 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
def normalize_to(value, actor):
    """Normalizes the to value

    ```pycon
    >>> normalize_to(None, "http://actor.example")
    ['http://actor.example']

    >>> normalize_to("http://to.example", None)
    ['http://to.example']

    >>> normalize_to(["http://alice.example", "http://bob.example"], None)
    ['http://alice.example', 'http://bob.example']

    ```
    """
    if value is None:
        return [actor]
    return list_from_value(value)

cattle_grid.activity_pub.muck_out.types

Activity

Bases: Common

This represents a first draft of a json-schema that every activities exchanged between servers MUST satisfy and be able to parse. Here ‘being able to parse’ means making it to the point, where depending on the type, you decide what side effects to perform.

Generally, the fields actor, to, and cc (and maybe bcc — not transported) represent how the message is being delivered. The fields actor, type, object, target, content represent how the message is processed by the server.

Source code in cattle_grid/activity_pub/muck_out/types.py 
124
125
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
class Activity(Common):
    """
    This represents a first draft of a json-schema that every activities exchanged between servers MUST satisfy and be able to parse. Here 'being able to parse' means making it to the point, where depending on the type, you decide what side effects to perform.

    Generally, the fields actor, to, and cc (and maybe bcc --- not transported) represent how the message is being delivered. The fields actor, type, object, target, content represent how the message is processed by the server.
    """

    actor: str = Field(
        ...,
        examples=["https://actor.example/"],
        description="""
    id of the actor performing this activity. One can assume that the activity is signed by this actor (in some form).
    """,
    )
    object: str | Object | None = Field(None)
    target: str | Dict[str, Any] | None = Field(
        None,
        examples=[
            "https://other.example/target_id",
            {"type": "Note", "content": "meow"},
        ],
        description="""
    The target, not sure if needed, included for completeness
    """,
    )
    content: str | None = Field(
        None,
        examples=["🐮", "❤️"],
        description="""
    The content used for example to represent the Emote for a like
    """,
    )
    id: str = Field(
        ...,
        examples=["https://actor.example/some_id"],
        description="""
    id of the activity, can be assumed to be globally unique. Some activities such as a Follow request will require an id to be valid. Servers may assume an id to be required. As assigning an id is 'trivial', one should assign one.
    """,
    )
    type: str = Field(
        ...,
        examples=["Follow", "Accept", "Create", "Undo", "Like"],
        description="""
    Type of the activity. Side effects of this activity are determine by this type.
    """,
    )
    to: List[str] = Field(
        ...,
        examples=[
            ["https://bob.example"],
            ["https://alice.example", "https://bob.example"],
        ],
        min_length=1,
        description="""
    Array of actors this activity is addressed to. It is sane to assume that an activity is addressed to at least one person.
    """,
    )

actor: str = Field(..., examples=['https://actor.example/'], description='\n id of the actor performing this activity. One can assume that the activity is signed by this actor (in some form).\n ') class-attribute instance-attribute

content: str | None = Field(None, examples=['🐮', '❤️'], description='\n The content used for example to represent the Emote for a like\n ') class-attribute instance-attribute

id: str = Field(..., examples=['https://actor.example/some_id'], description="\n id of the activity, can be assumed to be globally unique. Some activities such as a Follow request will require an id to be valid. Servers may assume an id to be required. As assigning an id is 'trivial', one should assign one.\n ") class-attribute instance-attribute

object: str | Object | None = Field(None) class-attribute instance-attribute

target: str | Dict[str, Any] | None = Field(None, examples=['https://other.example/target_id', {'type': 'Note', 'content': 'meow'}], description='\n The target, not sure if needed, included for completeness\n ') class-attribute instance-attribute

to: List[str] = Field(..., examples=[['https://bob.example'], ['https://alice.example', 'https://bob.example']], min_length=1, description='\n Array of actors this activity is addressed to. It is sane to assume that an activity is addressed to at least one person.\n ') class-attribute instance-attribute

type: str = Field(..., examples=['Follow', 'Accept', 'Create', 'Undo', 'Like'], description='\n Type of the activity. Side effects of this activity are determine by this type.\n ') class-attribute instance-attribute

Common

Bases: BaseModel

Source code in cattle_grid/activity_pub/muck_out/types.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
42
43
44
45
46
47
48
49
50
51
52
53
54
class Common(BaseModel):
    field_context: str | List[Any] | None = Field(
        None,
        alias="@context",
        examples=[
            "https://www.w3.org/ns/activitystreams",
            ["https://www.w3.org/ns/activitystreams", {"Hashtag": "as:Hashtag"}],
        ],
    )
    id: str = Field(
        ...,
        examples=["https://actor.example/some_id"],
        description="""
    id of the activity or object, can be assumed to be globally unique. Some activities such as a Follow request will require an id to be valid. Servers may assume an id to be required. As assigning an id is 'trivial', one should assign one.
    """,
    )
    to: List[str] = Field(
        ...,
        examples=[
            ["https://bob.example"],
            ["https://alice.example", "https://bob.example"],
        ],
        min_length=1,
        description="""
    Array of actors this activity or object is addressed to. It is sane to assume that an activity is addressed to at least one person.
    """,
    )
    cc: List[str] | None = Field(
        None,
        examples=[
            ["https://bob.example"],
            ["https://alice.example", "https://bob.example"],
        ],
        description="""
    Array of actors this activity or object is carbon copied to.
    """,
    )
    published: str | None = Field(
        None,
        description="""
    Moment of this activity or object being published
    """,
    )
    type: str = Field(
        ...,
        examples=["Follow", "Accept", "Create", "Undo", "Like", "Note"],
        description="""
    Type of the activity or activity. Side effects of this activity are determine by this type.
    """,
    )

cc: List[str] | None = Field(None, examples=[['https://bob.example'], ['https://alice.example', 'https://bob.example']], description='\n Array of actors this activity or object is carbon copied to.\n ') class-attribute instance-attribute

field_context: str | List[Any] | None = Field(None, alias='@context', examples=['https://www.w3.org/ns/activitystreams', ['https://www.w3.org/ns/activitystreams', {'Hashtag': 'as:Hashtag'}]]) class-attribute instance-attribute

id: str = Field(..., examples=['https://actor.example/some_id'], description="\n id of the activity or object, can be assumed to be globally unique. Some activities such as a Follow request will require an id to be valid. Servers may assume an id to be required. As assigning an id is 'trivial', one should assign one.\n ") class-attribute instance-attribute

published: str | None = Field(None, description='\n Moment of this activity or object being published\n ') class-attribute instance-attribute

to: List[str] = Field(..., examples=[['https://bob.example'], ['https://alice.example', 'https://bob.example']], min_length=1, description='\n Array of actors this activity or object is addressed to. It is sane to assume that an activity is addressed to at least one person.\n ') class-attribute instance-attribute

type: str = Field(..., examples=['Follow', 'Accept', 'Create', 'Undo', 'Like', 'Note'], description='\n Type of the activity or activity. Side effects of this activity are determine by this type.\n ') class-attribute instance-attribute

Object

Bases: Common

Source code in cattle_grid/activity_pub/muck_out/types.py 
 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
121
class Object(Common):
    attributedTo: str | None = Field(
        None,
        examples=["https://actor.example/"],
        description="""
    id of the actor that authored this object
    """,
    )
    content: str
    """
    The content of the object
    """
    summary: str | None = None
    """
    The summary of the object
    """
    name: str | None = None
    """
    The name of the object
    """
    attachment: List[Dict[str, Any]] | None = None
    """
    A list of objects that are attached to the original object
    """
    tag: List[Dict[str, Any]] | None = None
    """
    A list of objects that expand on the content of the object
    """
    url: List[Dict[str, Any] | str] | None = None
    """
    A list of urls that expand on the content of the object
    """
    sensitive: bool | None = None
    """
    Marks the object as sensitive. Currently, used by everyone, a better way would be an element of the tag list that labels the object as sensitive due a reason
    """
    inReplyTo: str | None = None
    """
    The object being replied to. Currently a string. Not sure if this is what I want.
    """
    id: str = Field(
        ...,
        examples=["https://actor.example/some_id"],
        description="""
    id of the activity, can be assumed to be globally unique. Some activities such as a Follow request will require an id to be valid. Servers may assume an id to be required. As assigning an id is 'trivial', one should assign one.
    """,
    )
    type: str = Field(
        ...,
        examples=["Follow", "Accept", "Create", "Undo", "Like"],
        description="""
    Type of the activity. Side effects of this activity are determine by this type.
    """,
    )
    to: List[str] = Field(
        ...,
        examples=[
            ["https://bob.example"],
            ["https://alice.example", "https://bob.example"],
        ],
        min_length=1,
        description="""
    Array of actors this activity is addressed to. It is sane to assume that an activity is addressed to at least one person.
    """,
    )

attachment: List[Dict[str, Any]] | None = None class-attribute instance-attribute

A list of objects that are attached to the original object

attributedTo: str | None = Field(None, examples=['https://actor.example/'], description='\n id of the actor that authored this object\n ') class-attribute instance-attribute

content: str instance-attribute

The content of the object

id: str = Field(..., examples=['https://actor.example/some_id'], description="\n id of the activity, can be assumed to be globally unique. Some activities such as a Follow request will require an id to be valid. Servers may assume an id to be required. As assigning an id is 'trivial', one should assign one.\n ") class-attribute instance-attribute

inReplyTo: str | None = None class-attribute instance-attribute

The object being replied to. Currently a string. Not sure if this is what I want.

name: str | None = None class-attribute instance-attribute

The name of the object

sensitive: bool | None = None class-attribute instance-attribute

Marks the object as sensitive. Currently, used by everyone, a better way would be an element of the tag list that labels the object as sensitive due a reason

summary: str | None = None class-attribute instance-attribute

The summary of the object

tag: List[Dict[str, Any]] | None = None class-attribute instance-attribute

A list of objects that expand on the content of the object

to: List[str] = Field(..., examples=[['https://bob.example'], ['https://alice.example', 'https://bob.example']], min_length=1, description='\n Array of actors this activity is addressed to. It is sane to assume that an activity is addressed to at least one person.\n ') class-attribute instance-attribute

type: str = Field(..., examples=['Follow', 'Accept', 'Create', 'Undo', 'Like'], description='\n Type of the activity. Side effects of this activity are determine by this type.\n ') class-attribute instance-attribute

url: List[Dict[str, Any] | str] | None = None class-attribute instance-attribute

A list of urls that expand on the content of the object