activity_tools.objects

  1import requests
  2import re
  3import uuid
  4from urllib.parse import urlparse
  5
  6from cryptography.hazmat.primitives.serialization import load_pem_public_key
  7
  8from .misc import ImageAsset, PublicKey, Tags, Attachment
  9
 10class Actor:
 11    """
 12    Generic actor object. Use `create(...)` to create an actor of your own,
 13    and `fetch(...)` to fetch an external actor.
 14    """
 15
 16    username: str
 17    """ The actors username"""
 18
 19    def __init__(self) -> None:
 20        """
 21        This creates an empty actor object
 22        """
 23        self.snake_pattern = re.compile(r'(?<!^)(?=[A-Z])')
 24
 25    def add_property_value(self, name, value) -> None:
 26        """
 27        Add a PropertyValue to the list of attachments. In Mastodon this
 28        is used for the links in the profile.
 29        """
 30        self.attachment.add_property_value(name, value)
 31
 32    def add_emoji(self, name, url) -> None:
 33        """
 34        Add a custom emoji to the tag list. If you for example like to
 35        map `:foo:` to `/images/foo.png`, this is the function for you!
 36        """
 37        self.tag.add_emoji(name, url)
 38
 39    def create(self, domain: str, username: str, public_key_bytes: bytes) -> None:
 40        """
 41        Populate the actor object with data. This is a useful to create the
 42        actor for a user. All fields can be overriden.
 43        """
 44        self.domain = domain
 45        self.username = username
 46        self.public_key = None
 47        self.public_key_pem = public_key_bytes.decode()
 48
 49        self.id = f"https://{self.domain}/users/{self.username}"
 50        self.type = "Person"
 51
 52        self.inbox = f"https://{self.domain}/users/{self.username}/inbox"
 53        self.outbox = f"https://{self.domain}/users/{self.username}/outbox"
 54
 55        self.following = f"https://{domain}/users/{username}/following"
 56        self.followers = f"https://{domain}/users/{username}/followers"
 57
 58        self.discoverable = False
 59        self.summary = ""
 60        self.published = "1523-06-06T10:00:00Z"
 61
 62        self.name = f"{self.username.capitalize()}"
 63        self.preferred_username = f"{self.username}"
 64
 65        self.icon_url = None
 66        self.image_url = None
 67        self.manually_approves_followers = None
 68        self.attachment = Attachment()
 69        self.tag = Tags(self.domain)
 70
 71    @classmethod
 72    def fetch(cls, actor_url):
 73        actor = Actor()
 74        actor._fetch(actor_url)
 75        return actor
 76
 77    def _fetch(self, actor_url):
 78        self.actor_raw = {}
 79
 80        if not actor_url:
 81            raise Exception("Actor URL is not set")
 82
 83        headers = {
 84            'Content-Type': "application/activity+json",
 85            'Accept': 'application/activity+json'
 86        }
 87
 88        actor_resp = requests.get(actor_url, headers=headers)
 89        if actor_resp.status_code > 299:
 90            raise Exception(f"Actor {actor_url} responded with a {actor_resp.status_code}")
 91
 92        self.actor_raw = actor_resp.json()
 93        urlid = urlparse(self.actor_raw['id'])
 94
 95        self.domain = urlid.netloc
 96
 97        keys = [
 98            "publicKey",
 99            "id",
100            "type",
101            "inbox",
102            "outbox",
103            "following",
104            "followers",
105            "discoverable",
106            "summary",
107            "published",
108            "name",
109            "preferredUsername",
110            "icon",
111            "image",
112            "manuallyApprovesFollowers",
113            "attachment",
114            "tag"
115        ]
116
117        for key in keys:
118            key_snake = self.snake_pattern.sub('_', key).lower()
119            setattr(self, key_snake, self.actor_raw.get(key))
120    
121        self.public_key_pem = self.public_key['publicKeyPem']
122
123    def get_public_key(self):
124        return load_pem_public_key(self.public_key_pem.encode())
125
126    def run(self) -> dict:
127        required_document = {
128            "@context": [
129                "https://www.w3.org/ns/activitystreams",
130                "https://w3id.org/security/v1",
131            ],
132            "id": self.id,
133            "type": self.type,
134            "inbox": self.inbox,
135            "discoverable": self.discoverable,
136            "summary": self.summary,
137            "published": self.published,
138            "name": self.name,
139            "preferredUsername": self.preferred_username,
140            "attachment": self.attachment.run(),
141            "tag": self.tag.run(),
142            "publicKey": PublicKey(
143                self.domain,
144                self.username,
145                self.public_key_pem
146            ).run(),
147        }
148
149        extra_values = {}
150
151        if self.icon_url:
152            extra_values["icon"] = ImageAsset(self.icon_url)
153
154        if self.icon_url:
155            extra_values["image"] = ImageAsset(self.image_url)
156
157        if self.manually_approves_followers:
158            extra_values["manuallyApprovesFollowers"] = self.manually_approves_followers
159
160        if self.followers:
161            extra_values["followers"] = self.followers
162
163        if self.following:
164            extra_values["following"] = self.following
165
166        if self.outbox:
167            extra_values["outbox"] = self.outbox
168
169        return { **required_document, **extra_values }
170
171
172class WrapActivityStreamsObject:
173
174    def __init__(self, object) -> None:
175        self.object = object
176
177    def run(self) -> dict:
178        context = {
179            "@context": "https://www.w3.org/ns/activitystreams"
180        }
181
182        return { **context, **self.object.run() }
183
184class InboxObject:
185
186    raw: dict
187    """
188    The raw data representing this object
189    """
190
191    def __init__(self, data) -> None:
192        self.raw = data
193        self.id = data['id']
194        self.type = data['type'].lower()
195        self._actor = data['actor']
196        self._object = data['object']
197
198    @property
199    def actor(self) -> Actor:
200        return Actor.fetch(actor_url=self._actor)
201
202    @property
203    def object(self) -> Actor:
204        return Actor.fetch(actor_url=self._object)
205
206    def run(self) -> dict:
207        """
208        Return the raw JSON data
209        """
210        return self.raw
211
212class Follow(InboxObject):
213
214    def __init__(self, data) -> None:
215        super().__init__(data)
216
217class Undo(InboxObject):
218
219    def __init__(self, data) -> None:
220        super().__init__(data)
221
222class ActivityPubObject:
223    
224    id: str
225    """
226    The objects ID, this should be a unique identifier
227    """
228
229    actor: Actor
230    """
231    The actor that created the message (the sender)
232    """
233
234    actor_url: str
235    """
236    A url to the actor that created the message (the sender)
237    """
238
239    object: object
240    """
241    A generic object representing the message/data. This may also
242    point to a target actor.
243    """
244
245    raw: dict
246    """
247    Object in raw unaltered form represented as a dict. This may be the
248    unaltered JSON document, but not always.
249    """
250
251class Accept(ActivityPubObject):
252
253    object: Follow
254    """ Object representing the Follow request """
255
256    def __init__(self, object: InboxObject) -> None:
257        self.id = f"{object.object.id}/accept/{str(uuid.uuid4())}"
258        self.object = object
259        self.raw = object.run()
260        self.actor = object.object
261
262    def run(self):
263        return {
264            "@context": "https://www.w3.org/ns/activitystreams",
265            "id": self.id,
266            "type": "Accept",
267            "actor": self.object.object.id,
268            "object": self.object.run()
269        }