kopia lustrzana https://github.com/snarfed/bridgy-fed
1810 wiersze
63 KiB
Python
1810 wiersze
63 KiB
Python
"""Datastore model classes."""
|
|
import copy
|
|
from datetime import timedelta, timezone
|
|
from functools import lru_cache
|
|
import itertools
|
|
import json
|
|
import logging
|
|
import random
|
|
import re
|
|
from threading import Lock
|
|
from urllib.parse import quote, urlparse
|
|
|
|
from arroba.util import parse_at_uri
|
|
import cachetools
|
|
from Crypto.PublicKey import RSA
|
|
from flask import request
|
|
from google.cloud import ndb
|
|
from granary import as1, as2, atom, bluesky, microformats2
|
|
from granary.bluesky import AT_URI_PATTERN, BSKY_APP_URL_RE
|
|
from granary.source import html_to_text
|
|
from oauth_dropins.webutil import util
|
|
from oauth_dropins.webutil.appengine_info import DEBUG
|
|
from oauth_dropins.webutil.flask_util import error
|
|
from oauth_dropins.webutil.models import JsonProperty, StringIdModel
|
|
from oauth_dropins.webutil.util import ellipsize, json_dumps, json_loads
|
|
from requests import RequestException
|
|
|
|
import common
|
|
from common import (
|
|
base64_to_long,
|
|
DOMAIN_RE,
|
|
long_to_base64,
|
|
OLD_ACCOUNT_AGE,
|
|
PROTOCOL_DOMAINS,
|
|
report_error,
|
|
unwrap,
|
|
)
|
|
import ids
|
|
import memcache
|
|
|
|
# maps string label to Protocol subclass. values are populated by ProtocolUserMeta.
|
|
# (we used to wait for ProtocolUserMeta to populate the keys as well, but that was
|
|
# awkward to use in datastore model properties with choices, below; it required
|
|
# overriding them in reset_model_properties, which was always flaky.)
|
|
PROTOCOLS = {label: None for label in (
|
|
'activitypub',
|
|
'ap',
|
|
'atproto',
|
|
'bsky',
|
|
'ostatus',
|
|
'web',
|
|
'webmention',
|
|
'ui',
|
|
)}
|
|
if DEBUG:
|
|
PROTOCOLS.update({label: None for label in (
|
|
'fa',
|
|
'fake',
|
|
'efake',
|
|
'other',
|
|
)})
|
|
|
|
# maps string kind (eg 'MagicKey') to Protocol subclass.
|
|
# populated in ProtocolUserMeta
|
|
PROTOCOLS_BY_KIND = {}
|
|
|
|
|
|
# 2048 bits makes tests slow, so use 1024 for them
|
|
KEY_BITS = 1024 if DEBUG else 2048
|
|
PAGE_SIZE = 20
|
|
|
|
# auto delete most old objects via the Object.expire property
|
|
# https://cloud.google.com/datastore/docs/ttl
|
|
DONT_EXPIRE_OBJECT_TYPES = \
|
|
as1.ACTOR_TYPES | as1.POST_TYPES | set(('event', 'question'))
|
|
OBJECT_EXPIRE_AGE = timedelta(days=90)
|
|
|
|
GET_ORIGINALS_CACHE_EXPIRATION = timedelta(days=1)
|
|
FOLLOWERS_CACHE_EXPIRATION = timedelta(hours=2)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class Target(ndb.Model):
|
|
r""":class:`protocol.Protocol` + URI pairs for identifying objects.
|
|
|
|
These are currently used for:
|
|
|
|
* delivery destinations, eg ActivityPub inboxes, webmention targets, etc.
|
|
* copies of :class:`Object`\s and :class:`User`\s elsewhere,
|
|
eg ``at://`` URIs for ATProto records, nevent etc bech32-encoded Nostr ids,
|
|
ATProto user DIDs, etc.
|
|
|
|
Used in :class:`google.cloud.ndb.model.StructuredProperty`\s inside
|
|
:class:`Object` and :class:`User`; not stored as top-level entities in the
|
|
datastore.
|
|
|
|
ndb implements this by hoisting each property here into a corresponding
|
|
property on the parent entity, prefixed by the StructuredProperty name
|
|
below, eg ``delivered.uri``, ``delivered.protocol``, etc.
|
|
|
|
For repeated StructuredPropertys, the hoisted properties are all repeated on
|
|
the parent entity, and reconstructed into StructuredPropertys based on their
|
|
order.
|
|
|
|
https://googleapis.dev/python/python-ndb/latest/model.html#google.cloud.ndb.model.StructuredProperty
|
|
"""
|
|
uri = ndb.StringProperty(required=True)
|
|
protocol = ndb.StringProperty(choices=list(PROTOCOLS.keys()), required=True)
|
|
|
|
def __eq__(self, other):
|
|
"""Equality excludes Targets' :class:`Key`."""
|
|
return self.uri == other.uri and self.protocol == other.protocol
|
|
|
|
def __hash__(self):
|
|
"""Allow hashing so these can be dict keys."""
|
|
return hash((self.protocol, self.uri))
|
|
|
|
|
|
class DM(ndb.Model):
|
|
""":class:`protocol.Protocol` + type pairs for identifying sent DMs.
|
|
|
|
Used in :attr:`User.sent_dms`.
|
|
|
|
https://googleapis.dev/python/python-ndb/latest/model.html#google.cloud.ndb.model.StructuredProperty
|
|
"""
|
|
TYPES = (
|
|
'request_bridging',
|
|
'replied_to_bridged_user',
|
|
'welcome',
|
|
)
|
|
type = ndb.StringProperty(choices=TYPES, required=True)
|
|
protocol = ndb.StringProperty(choices=list(PROTOCOLS.keys()), required=True)
|
|
|
|
def __eq__(self, other):
|
|
"""Equality excludes Targets' :class:`Key`."""
|
|
return self.type == other.type and self.protocol == other.protocol
|
|
|
|
|
|
class ProtocolUserMeta(type(ndb.Model)):
|
|
""":class:`User` metaclass. Registers all subclasses in the ``PROTOCOLS`` global."""
|
|
def __new__(meta, name, bases, class_dict):
|
|
cls = super().__new__(meta, name, bases, class_dict)
|
|
|
|
if hasattr(cls, 'LABEL') and cls.LABEL not in ('protocol', 'user'):
|
|
for label in (cls.LABEL, cls.ABBREV) + cls.OTHER_LABELS:
|
|
if label:
|
|
PROTOCOLS[label] = cls
|
|
|
|
PROTOCOLS_BY_KIND[cls._get_kind()] = cls
|
|
|
|
return cls
|
|
|
|
|
|
def reset_protocol_properties():
|
|
"""Recreates various protocol properties to include choices from ``PROTOCOLS``."""
|
|
abbrevs = f'({"|".join(PROTOCOLS.keys())}|fed)'
|
|
common.SUBDOMAIN_BASE_URL_RE = re.compile(
|
|
rf'^https?://({abbrevs}\.brid\.gy|localhost(:8080)?)/(convert/|r/)?({abbrevs}/)?(?P<path>.+)')
|
|
ids.COPIES_PROTOCOLS = tuple(label for label, proto in PROTOCOLS.items()
|
|
if proto and proto.HAS_COPIES)
|
|
|
|
|
|
class User(StringIdModel, metaclass=ProtocolUserMeta):
|
|
"""Abstract base class for a Bridgy Fed user.
|
|
|
|
Stores some protocols' keypairs. Currently:
|
|
|
|
* RSA keypair for ActivityPub HTTP Signatures
|
|
properties: ``mod``, ``public_exponent``, ``private_exponent``, all
|
|
encoded as base64url (ie URL-safe base64) strings as described in RFC
|
|
4648 and section 5.1 of the Magic Signatures spec:
|
|
https://tools.ietf.org/html/draft-cavage-http-signatures-12
|
|
* *Not* K-256 signing or rotation keys for AT Protocol, those are stored in
|
|
:class:`arroba.datastore_storage.AtpRepo` entities
|
|
"""
|
|
obj_key = ndb.KeyProperty(kind='Object') # user profile
|
|
mod = ndb.StringProperty()
|
|
use_instead = ndb.KeyProperty()
|
|
|
|
# Proxy copies of this user elsewhere, eg DIDs for ATProto records, bech32
|
|
# npub Nostr ids, etc. Similar to rel-me links in microformats2, alsoKnownAs
|
|
# in DID docs (and now AS2), etc.
|
|
# TODO: switch to using Object.copies on the user profile object?
|
|
copies = ndb.StructuredProperty(Target, repeated=True)
|
|
|
|
# these are for ActivityPub HTTP Signatures
|
|
public_exponent = ndb.StringProperty()
|
|
private_exponent = ndb.StringProperty()
|
|
|
|
# set to True for users who asked me to be opted out instead of putting
|
|
# #nobridge in their profile
|
|
manual_opt_out = ndb.BooleanProperty()
|
|
|
|
# protocols that this user has explicitly opted into. protocols that don't
|
|
# require explicit opt in are omitted here. choices is populated in
|
|
# reset_protocol_properties.
|
|
enabled_protocols = ndb.StringProperty(repeated=True, choices=list(PROTOCOLS.keys()))
|
|
|
|
# DMs that we've attempted to send to this user
|
|
sent_dms = ndb.StructuredProperty(DM, repeated=True)
|
|
|
|
created = ndb.DateTimeProperty(auto_now_add=True)
|
|
updated = ndb.DateTimeProperty(auto_now=True)
|
|
|
|
# `existing` attr is set by get_or_create
|
|
|
|
# OLD. some stored entities still have these; do not reuse.
|
|
# direct = ndb.BooleanProperty(default=False)
|
|
# actor_as2 = JsonProperty()
|
|
# protocol-specific state
|
|
# atproto_notifs_indexed_at = ndb.TextProperty()
|
|
# atproto_feed_indexed_at = ndb.TextProperty()
|
|
|
|
def __init__(self, **kwargs):
|
|
"""Constructor.
|
|
|
|
Sets :attr:`obj` explicitly because however
|
|
:class:`google.cloud.ndb.model.Model` sets it doesn't work with
|
|
``@property`` and ``@obj.setter`` below.
|
|
"""
|
|
obj = kwargs.pop('obj', None)
|
|
super().__init__(**kwargs)
|
|
|
|
if obj:
|
|
self.obj = obj
|
|
|
|
self.lock = Lock()
|
|
|
|
@classmethod
|
|
def new(cls, **kwargs):
|
|
"""Try to prevent instantiation. Use subclasses instead."""
|
|
raise NotImplementedError()
|
|
|
|
def _post_put_hook(self, future):
|
|
logger.debug(f'Wrote {self.key}')
|
|
|
|
def add(self, prop, val):
|
|
"""Adds a value to a multiply-valued property. Uses ``self.lock``.
|
|
|
|
Args:
|
|
prop (str)
|
|
val
|
|
"""
|
|
with self.lock:
|
|
added = util.add(getattr(self, prop), val)
|
|
|
|
if prop == 'copies' and added:
|
|
memcache.pickle_memcache.set(memcache.memoize_key(
|
|
get_original_user_key, val.uri), self.key)
|
|
|
|
@classmethod
|
|
def get_by_id(cls, id, allow_opt_out=False, **kwargs):
|
|
"""Override to follow ``use_instead`` property and ``opt-out` status.
|
|
|
|
Returns None if the user is opted out.
|
|
"""
|
|
user = cls._get_by_id(id, **kwargs)
|
|
if user and user.use_instead:
|
|
logger.info(f'{user.key} use_instead => {user.use_instead}')
|
|
user = user.use_instead.get()
|
|
|
|
if not user:
|
|
return None
|
|
|
|
if user.status and not allow_opt_out:
|
|
logger.info(f'{user.key} is {user.status}')
|
|
return None
|
|
|
|
return user
|
|
|
|
@classmethod
|
|
def get_or_create(cls, id, propagate=False, allow_opt_out=False,
|
|
reload=False, **kwargs):
|
|
"""Loads and returns a :class:`User`. Creates it if necessary.
|
|
|
|
Not transactional because transactions don't read or write memcache. :/
|
|
Fortunately we don't really depend on atomicity for anything, last
|
|
writer wins is pretty much always fine.
|
|
|
|
Args:
|
|
propagate (bool): whether to create copies of this user in push-based
|
|
protocols, eg ATProto and Nostr.
|
|
allow_opt_out (bool): whether to allow and create the user if they're
|
|
currently opted out
|
|
reload (bool): whether to reload profile always, vs only if necessary
|
|
kwargs: passed through to ``cls`` constructor
|
|
|
|
Returns:
|
|
User: existing or new user, or None if the user is opted out
|
|
"""
|
|
assert cls != User
|
|
|
|
user = cls.get_by_id(id, allow_opt_out=True)
|
|
if user:
|
|
if reload:
|
|
user.reload_profile(gateway=True, raise_=False)
|
|
|
|
if user.status and not allow_opt_out:
|
|
return None
|
|
user.existing = True
|
|
|
|
# TODO: propagate more fields?
|
|
changed = False
|
|
for field in ['obj', 'obj_key']:
|
|
old_val = getattr(user, field, None)
|
|
new_val = kwargs.get(field)
|
|
if old_val is None and new_val is not None:
|
|
setattr(user, field, new_val)
|
|
changed = True
|
|
|
|
if enabled_protocols := kwargs.get('enabled_protocols'):
|
|
user.enabled_protocols = (set(user.enabled_protocols)
|
|
| set(enabled_protocols))
|
|
changed = True
|
|
|
|
if not propagate:
|
|
if changed:
|
|
user.put()
|
|
return user
|
|
|
|
else:
|
|
if orig_key := get_original_user_key(id):
|
|
orig = orig_key.get()
|
|
if orig.status and not allow_opt_out:
|
|
return None
|
|
orig.existing = False
|
|
return orig
|
|
|
|
user = cls(id=id, **kwargs)
|
|
user.existing = False
|
|
user.reload_profile(gateway=True, raise_=False)
|
|
if user.status and not allow_opt_out:
|
|
return None
|
|
|
|
if propagate and not user.status:
|
|
for label in user.enabled_protocols + list(user.DEFAULT_ENABLED_PROTOCOLS):
|
|
proto = PROTOCOLS[label]
|
|
if proto == cls:
|
|
continue
|
|
elif proto.HAS_COPIES:
|
|
if not user.get_copy(proto) and user.is_enabled(proto):
|
|
try:
|
|
proto.create_for(user)
|
|
except (ValueError, AssertionError):
|
|
logger.info(f'failed creating {proto.LABEL} copy',
|
|
exc_info=True)
|
|
util.remove(user.enabled_protocols, proto.LABEL)
|
|
else:
|
|
logger.debug(f'{proto.LABEL} not enabled or user copy already exists, skipping propagate')
|
|
|
|
# generate keys for all protocols _except_ our own
|
|
#
|
|
# these can use urandom() and do nontrivial math, so they can take time
|
|
# depending on the amount of randomness available and compute needed.
|
|
if cls.LABEL != 'activitypub':
|
|
if (not user.public_exponent or not user.private_exponent or not user.mod):
|
|
assert (not user.public_exponent and not user.private_exponent
|
|
and not user.mod), id
|
|
key = RSA.generate(KEY_BITS,
|
|
randfunc=random.randbytes if DEBUG else None)
|
|
user.mod = long_to_base64(key.n)
|
|
user.public_exponent = long_to_base64(key.e)
|
|
user.private_exponent = long_to_base64(key.d)
|
|
|
|
try:
|
|
user.put()
|
|
except AssertionError as e:
|
|
error(f'Bad {cls.__name__} id {id} : {e}')
|
|
|
|
logger.debug(('Updated ' if user.existing else 'Created new ') + str(user))
|
|
return user
|
|
|
|
@property
|
|
def obj(self):
|
|
"""Convenience accessor that loads :attr:`obj_key` from the datastore."""
|
|
if self.obj_key:
|
|
if not hasattr(self, '_obj'):
|
|
self._obj = self.obj_key.get()
|
|
return self._obj
|
|
|
|
@obj.setter
|
|
def obj(self, obj):
|
|
if obj:
|
|
assert isinstance(obj, Object)
|
|
assert obj.key
|
|
self._obj = obj
|
|
self.obj_key = obj.key
|
|
else:
|
|
self._obj = self.obj_key = None
|
|
|
|
def delete(self, proto=None):
|
|
"""Deletes a user's bridged actors in all protocols or a specific one.
|
|
|
|
Args:
|
|
proto (Protocol): optional
|
|
"""
|
|
now = util.now().isoformat()
|
|
proto_label = proto.LABEL if proto else 'all'
|
|
delete_id = f'{self.profile_id()}#bridgy-fed-delete-user-{proto_label}-{now}'
|
|
delete = Object(id=delete_id, source_protocol=self.LABEL, our_as1={
|
|
'id': delete_id,
|
|
'objectType': 'activity',
|
|
'verb': 'delete',
|
|
'actor': self.key.id(),
|
|
'object': self.key.id(),
|
|
})
|
|
delete.put()
|
|
self.deliver(delete, from_user=self, to_proto=proto)
|
|
|
|
@classmethod
|
|
def load_multi(cls, users):
|
|
"""Loads :attr:`obj` for multiple users in parallel.
|
|
|
|
Args:
|
|
users (sequence of User)
|
|
"""
|
|
objs = ndb.get_multi(u.obj_key for u in users if u.obj_key)
|
|
keys_to_objs = {o.key: o for o in objs if o}
|
|
|
|
for u in users:
|
|
u._obj = keys_to_objs.get(u.obj_key)
|
|
|
|
@ndb.ComputedProperty
|
|
def handle(self):
|
|
"""This user's unique, human-chosen handle, eg ``@me@snarfed.org``.
|
|
|
|
To be implemented by subclasses.
|
|
"""
|
|
raise NotImplementedError()
|
|
|
|
@ndb.ComputedProperty
|
|
def readable_id(self):
|
|
"""DEPRECATED: replaced by handle. Kept for backward compatibility."""
|
|
return None
|
|
|
|
@ndb.ComputedProperty
|
|
def status(self):
|
|
"""Whether this user is blocked or opted out.
|
|
|
|
Optional. Current possible values:
|
|
* ``opt-out``: if ``#nobridge`` or ``#nobot`` is in the profile
|
|
description/bio, or if the user or domain has manually opted out.
|
|
Some protocols also have protocol-specific opt out logic, eg Bluesky
|
|
accounts that have disabled logged out view.
|
|
* ``blocked``: if the user fails our validation checks, eg
|
|
``REQUIRES_NAME`` or ``REQUIRES_AVATAR`` if either of those are
|
|
``True` for this protocol.
|
|
* `owns-webfinger`: a :class:`web.Web` user that looks like a
|
|
fediverse server
|
|
* `no-feed-or-webmention`: a :class:`web.Web` user that doesn't have
|
|
an RSS or Atom feed or webmention endpoint and has never sent us a
|
|
webmention
|
|
|
|
Duplicates ``util.is_opt_out`` in Bridgy!
|
|
|
|
https://github.com/snarfed/bridgy-fed/issues/666
|
|
"""
|
|
if self.manual_opt_out:
|
|
return 'opt-out'
|
|
|
|
if not self.obj or not self.obj.as1:
|
|
return None
|
|
|
|
if self.REQUIRES_AVATAR and not self.obj.as1.get('image'):
|
|
return 'blocked'
|
|
|
|
name = self.obj.as1.get('displayName')
|
|
if self.REQUIRES_NAME and (not name or name in (self.handle, self.key.id())):
|
|
return 'blocked'
|
|
|
|
if self.REQUIRES_OLD_ACCOUNT:
|
|
if published := self.obj.as1.get('published'):
|
|
if util.now() - util.parse_iso8601(published) < OLD_ACCOUNT_AGE:
|
|
return 'blocked'
|
|
|
|
summary = html_to_text(self.obj.as1.get('summary', ''), ignore_links=True)
|
|
name = self.obj.as1.get('displayName', '')
|
|
|
|
# #nobridge overrides enabled_protocols
|
|
if '#nobridge' in summary or '#nobridge' in name:
|
|
return 'opt-out'
|
|
|
|
# user has explicitly opted in. should go after spam filter (REQUIRES_*)
|
|
# checks, but before is_public and #nobot
|
|
if self.enabled_protocols:
|
|
return None
|
|
|
|
if not as1.is_public(self.obj.as1, unlisted=False):
|
|
return 'opt-out'
|
|
|
|
# enabled_protocols overrides #nobot
|
|
if '#nobot' in summary or '#nobot' in name:
|
|
return 'opt-out'
|
|
|
|
def is_enabled(self, to_proto, explicit=False):
|
|
"""Returns True if this user can be bridged to a given protocol.
|
|
|
|
Reasons this might return False:
|
|
* We haven't turned on bridging these two protocols yet.
|
|
* The user is opted out or blocked.
|
|
* The user is on a domain that's opted out or blocked.
|
|
* The from protocol requires opt in, and the user hasn't opted in.
|
|
* ``explicit`` is True, and this protocol supports ``to_proto`` by
|
|
default, but the user hasn't explicitly opted into it.
|
|
|
|
Args:
|
|
to_proto (Protocol subclass)
|
|
explicit (bool)
|
|
|
|
Returns:
|
|
bool:
|
|
"""
|
|
from protocol import Protocol
|
|
assert issubclass(to_proto, Protocol)
|
|
|
|
if self.__class__ == to_proto:
|
|
return True
|
|
|
|
from_label = self.LABEL
|
|
to_label = to_proto.LABEL
|
|
|
|
if bot_protocol := Protocol.for_bridgy_subdomain(self.key.id()):
|
|
return to_proto != bot_protocol
|
|
|
|
elif self.manual_opt_out:
|
|
return False
|
|
|
|
elif to_label in self.enabled_protocols:
|
|
return True
|
|
|
|
elif self.status:
|
|
return False
|
|
|
|
elif to_label in self.DEFAULT_ENABLED_PROTOCOLS and not explicit:
|
|
return True
|
|
|
|
return False
|
|
|
|
def enable_protocol(self, to_proto):
|
|
"""Adds ``to_proto` to :attr:`enabled_protocols`.
|
|
|
|
Also sends a welcome DM to the user (via a send task) if their protocol
|
|
supports DMs.
|
|
|
|
Args:
|
|
to_proto (:class:`protocol.Protocol` subclass)
|
|
"""
|
|
added = False
|
|
|
|
if to_proto.LABEL in ids.COPIES_PROTOCOLS:
|
|
# do this even if there's an existing copy since we might need to
|
|
# reactivate it, which create_for should do
|
|
to_proto.create_for(self)
|
|
|
|
@ndb.transactional()
|
|
def enable():
|
|
user = self.key.get()
|
|
if to_proto.LABEL not in user.enabled_protocols:
|
|
user.enabled_protocols.append(to_proto.LABEL)
|
|
util.add(user.sent_dms, DM(protocol=to_proto.LABEL, type='welcome'))
|
|
user.put()
|
|
nonlocal added
|
|
added = True
|
|
|
|
return user
|
|
|
|
new_self = enable()
|
|
# populate newly enabled protocol in this instance
|
|
self.enabled_protocols = new_self.enabled_protocols
|
|
self.copies = new_self.copies
|
|
if self.obj:
|
|
self.obj.copies = new_self.obj.copies
|
|
|
|
if added:
|
|
import dms
|
|
dms.maybe_send(from_proto=to_proto, to_user=self, type='welcome',
|
|
text=f"""\
|
|
Welcome to Bridgy Fed! Your account will soon be bridged to {to_proto.PHRASE} at {self.user_link(proto=to_proto, name=False)}. <a href="https://fed.brid.gy/docs">See the docs</a> and <a href="https://{common.PRIMARY_DOMAIN}{self.user_page_path()}">your user page</a> for more information. To disable this and delete your bridged profile, block this account.""")
|
|
|
|
msg = f'Enabled {to_proto.LABEL} for {self.key.id()} : {self.user_page_path()}'
|
|
logger.info(msg)
|
|
|
|
def disable_protocol(self, to_proto):
|
|
"""Removes ``to_proto` from :attr:`enabled_protocols`.
|
|
|
|
Args:
|
|
to_proto (:class:`protocol.Protocol` subclass)
|
|
"""
|
|
@ndb.transactional()
|
|
def disable():
|
|
user = self.key.get()
|
|
util.remove(user.enabled_protocols, to_proto.LABEL)
|
|
user.put()
|
|
|
|
disable()
|
|
util.remove(self.enabled_protocols, to_proto.LABEL)
|
|
|
|
msg = f'Disabled {to_proto.LABEL} for {self.key.id()} : {self.user_page_path()}'
|
|
logger.info(msg)
|
|
|
|
def handle_as(self, to_proto):
|
|
"""Returns this user's handle in a different protocol.
|
|
|
|
Args:
|
|
to_proto (str or Protocol)
|
|
|
|
Returns:
|
|
str
|
|
"""
|
|
if isinstance(to_proto, str):
|
|
to_proto = PROTOCOLS[to_proto]
|
|
|
|
# override to-ATProto to use custom domain handle in DID doc
|
|
from atproto import ATProto, did_to_handle
|
|
if to_proto == ATProto:
|
|
if did := self.get_copy(ATProto):
|
|
if handle := did_to_handle(did, remote=False):
|
|
return handle
|
|
|
|
# override web users to always use domain instead of custom username
|
|
# TODO: fall back to id if handle is unset?
|
|
handle = self.key.id() if self.LABEL == 'web' else self.handle
|
|
if not handle:
|
|
return None
|
|
|
|
return ids.translate_handle(handle=handle, from_=self.__class__,
|
|
to=to_proto, enhanced=False)
|
|
|
|
def id_as(self, to_proto):
|
|
"""Returns this user's id in a different protocol.
|
|
|
|
Args:
|
|
to_proto (str or Protocol)
|
|
|
|
Returns:
|
|
str
|
|
"""
|
|
if isinstance(to_proto, str):
|
|
to_proto = PROTOCOLS[to_proto]
|
|
|
|
return ids.translate_user_id(id=self.key.id(), from_=self.__class__,
|
|
to=to_proto)
|
|
|
|
def handle_or_id(self):
|
|
"""Returns handle if we know it, otherwise id."""
|
|
return self.handle or self.key.id()
|
|
|
|
def public_pem(self):
|
|
"""
|
|
Returns:
|
|
bytes:
|
|
"""
|
|
rsa = RSA.construct((base64_to_long(str(self.mod)),
|
|
base64_to_long(str(self.public_exponent))))
|
|
return rsa.exportKey(format='PEM')
|
|
|
|
def private_pem(self):
|
|
"""
|
|
Returns:
|
|
bytes:
|
|
"""
|
|
assert self.mod and self.public_exponent and self.private_exponent, str(self)
|
|
rsa = RSA.construct((base64_to_long(str(self.mod)),
|
|
base64_to_long(str(self.public_exponent)),
|
|
base64_to_long(str(self.private_exponent))))
|
|
return rsa.exportKey(format='PEM')
|
|
|
|
def name(self):
|
|
"""Returns this user's human-readable name, eg ``Ryan Barrett``."""
|
|
if self.obj and self.obj.as1:
|
|
name = self.obj.as1.get('displayName')
|
|
if name:
|
|
return name
|
|
|
|
return self.handle_or_id()
|
|
|
|
def web_url(self):
|
|
"""Returns this user's web URL (homepage), eg ``https://foo.com/``.
|
|
|
|
To be implemented by subclasses.
|
|
|
|
Returns:
|
|
str
|
|
"""
|
|
raise NotImplementedError()
|
|
|
|
def is_web_url(self, url, ignore_www=False):
|
|
"""Returns True if the given URL is this user's web URL (homepage).
|
|
|
|
Args:
|
|
url (str)
|
|
ignore_www (bool): if True, ignores ``www.`` subdomains
|
|
|
|
Returns:
|
|
bool:
|
|
"""
|
|
if not url:
|
|
return False
|
|
|
|
url = url.strip().rstrip('/')
|
|
url = re.sub(r'^(https?://)www\.', r'\1', url)
|
|
parsed_url = urlparse(url)
|
|
if parsed_url.scheme not in ('http', 'https', ''):
|
|
return False
|
|
|
|
this = self.web_url().rstrip('/')
|
|
this = re.sub(r'^(https?://)www\.', r'\1', this)
|
|
parsed_this = urlparse(this)
|
|
|
|
return (url == this or url == parsed_this.netloc or
|
|
parsed_url[1:] == parsed_this[1:]) # ignore http vs https
|
|
|
|
def id_uri(self):
|
|
"""Returns the user id as a URI.
|
|
|
|
Sometimes this is the user id itself, eg ActivityPub actor ids.
|
|
Sometimes it's a bit different, eg at://did:plc:... for ATProto user,
|
|
https://site.com for Web users.
|
|
|
|
Returns:
|
|
str
|
|
"""
|
|
return self.key.id()
|
|
|
|
def profile_id(self):
|
|
"""Returns the id of this user's profile object in its native protocol.
|
|
|
|
Examples:
|
|
|
|
* Web: home page URL, eg ``https://me.com/``
|
|
* ActivityPub: actor URL, eg ``https://instance.com/users/me``
|
|
* ATProto: profile AT URI, eg ``at://did:plc:123/app.bsky.actor.profile/self``
|
|
|
|
Defaults to this user's key id.
|
|
|
|
Returns:
|
|
str or None:
|
|
"""
|
|
return ids.profile_id(id=self.key.id(), proto=self)
|
|
|
|
def reload_profile(self, **kwargs):
|
|
"""Reloads this user's identity and profile from their native protocol.
|
|
|
|
Populates the reloaded profile :class:`Object` in ``self.obj``.
|
|
|
|
Args:
|
|
kwargs: passed through to :meth:`Protocol.load`
|
|
"""
|
|
obj = self.load(self.profile_id(), remote=True, **kwargs)
|
|
if obj:
|
|
self.obj = obj
|
|
|
|
def user_page_path(self, rest=None):
|
|
"""Returns the user's Bridgy Fed user page path."""
|
|
path = f'/{self.ABBREV}/{self.handle_or_id()}'
|
|
|
|
if rest:
|
|
if not rest.startswith('?'):
|
|
path += '/'
|
|
path += rest
|
|
|
|
return path
|
|
|
|
def get_copy(self, proto):
|
|
"""Returns the id for the copy of this user in a given protocol.
|
|
|
|
...or None if no such copy exists. If ``proto`` is this user, returns
|
|
this user's key id.
|
|
|
|
Args:
|
|
proto: :class:`Protocol` subclass
|
|
|
|
Returns:
|
|
str:
|
|
"""
|
|
# don't use isinstance because the testutil Fake protocol has subclasses
|
|
if self.LABEL == proto.LABEL:
|
|
return self.key.id()
|
|
|
|
for copy in self.copies:
|
|
if copy.protocol in (proto.LABEL, proto.ABBREV):
|
|
return copy.uri
|
|
|
|
def user_link(self, name=True, handle=True, pictures=False, proto=None,
|
|
proto_fallback=False):
|
|
"""Returns a pretty HTML link to the user's profile.
|
|
|
|
Can optionally include display name, handle, profile
|
|
picture, and/or link to a different protocol that they've enabled.
|
|
|
|
TODO: unify with :meth:`Object.actor_link`?
|
|
|
|
Args:
|
|
name (bool): include display name
|
|
handle (bool): include handle
|
|
pictures (bool): include profile picture and protocol logo
|
|
proto (protocol.Protocol): link to this protocol instead of the user's
|
|
native protocol
|
|
proto_fallback (bool): if True, and ``proto`` is provided and has no
|
|
no canonical profile URL for bridged users, uses the user's profile
|
|
URL in their native protocol
|
|
"""
|
|
img = name_str = handle_str = dot = logo = a_open = a_close = ''
|
|
|
|
if proto:
|
|
assert self.is_enabled(proto), f"{proto.LABEL} isn't enabled"
|
|
url = proto.bridged_web_url_for(self, fallback=proto_fallback)
|
|
else:
|
|
proto = self.__class__
|
|
url = self.web_url()
|
|
|
|
if pictures:
|
|
logo = f'<span class="logo" title="{proto.__name__}">{proto.LOGO_HTML}</span> '
|
|
if pic := self.profile_picture():
|
|
img = f'<img src="{pic}" class="profile"> '
|
|
|
|
if handle:
|
|
handle_str = self.handle_as(proto) or ''
|
|
|
|
if name and self.name() != handle_str:
|
|
name_str = self.name() or ''
|
|
|
|
if handle_str and name_str:
|
|
dot = ' · '
|
|
|
|
if url:
|
|
a_open = f'<a class="h-card u-author" rel="me" href="{url}" title="{name_str}{dot}{handle_str}">'
|
|
a_close = '</a>'
|
|
|
|
name_html = f'<span style="unicode-bidi: isolate">{ellipsize(name_str, chars=40)}</span>' if name_str else ''
|
|
return f'{logo}{a_open}{img}{name_html}{dot}{ellipsize(handle_str, chars=40)}{a_close}'
|
|
|
|
def profile_picture(self):
|
|
"""Returns the user's profile picture image URL, if available, or None."""
|
|
if self.obj and self.obj.as1:
|
|
return util.get_url(self.obj.as1, 'image')
|
|
|
|
# can't use functools.lru_cache here because we want the cache key to be
|
|
# just the user id, not the whole entity
|
|
@cachetools.cached(
|
|
cachetools.TTLCache(50000, FOLLOWERS_CACHE_EXPIRATION.total_seconds()),
|
|
key=lambda user: user.key.id(), lock=Lock())
|
|
@memcache.memoize(key=lambda self: self.key.id(),
|
|
expire=FOLLOWERS_CACHE_EXPIRATION)
|
|
def count_followers(self):
|
|
"""Counts this user's followers and followings.
|
|
|
|
Returns:
|
|
(int, int) tuple: (number of followers, number following)
|
|
"""
|
|
if self.key.id() in PROTOCOL_DOMAINS:
|
|
# we don't store Followers for protocol bot users any more, so
|
|
# follower counts are inaccurate, so don't return them
|
|
return (0, 0)
|
|
|
|
num_followers = Follower.query(Follower.to == self.key,
|
|
Follower.status == 'active')\
|
|
.count_async()
|
|
num_following = Follower.query(Follower.from_ == self.key,
|
|
Follower.status == 'active')\
|
|
.count_async()
|
|
return num_followers.get_result(), num_following.get_result()
|
|
|
|
|
|
class Object(StringIdModel):
|
|
"""An activity or other object, eg actor.
|
|
|
|
Key name is the id. We synthesize ids if necessary.
|
|
"""
|
|
LABELS = ('activity',
|
|
# DEPRECATED, replaced by users, notify, feed
|
|
'feed', 'notification', 'user')
|
|
|
|
# Keys for user(s) who created or otherwise own this activity.
|
|
users = ndb.KeyProperty(repeated=True)
|
|
# User keys who should see this activity in their user page, eg in reply to,
|
|
# reaction to, share of, etc.
|
|
notify = ndb.KeyProperty(repeated=True)
|
|
# User keys who should see this activity in their feeds, eg followers of its
|
|
# creator
|
|
feed = ndb.KeyProperty(repeated=True)
|
|
|
|
# DEPRECATED but still used read only to maintain backward compatibility
|
|
# with old Objects in the datastore that we haven't bothered migrating.
|
|
domains = ndb.StringProperty(repeated=True)
|
|
|
|
# choices is populated in reset_protocol_properties, after all User
|
|
# subclasses are created, so that PROTOCOLS is fully populated.
|
|
# TODO: nail down whether this is ABBREV or LABEL
|
|
source_protocol = ndb.StringProperty(choices=list(PROTOCOLS.keys()))
|
|
labels = ndb.StringProperty(repeated=True, choices=LABELS)
|
|
|
|
# TODO: switch back to ndb.JsonProperty if/when they fix it for the web console
|
|
# https://github.com/googleapis/python-ndb/issues/874
|
|
as2 = JsonProperty() # only one of the rest will be populated...
|
|
bsky = JsonProperty() # Bluesky / AT Protocol
|
|
mf2 = JsonProperty() # HTML microformats2 item (ie _not_ the top level
|
|
# parse object with items inside an 'items' field)
|
|
our_as1 = JsonProperty() # AS1 for activities that we generate or modify ourselves
|
|
raw = JsonProperty() # other standalone data format, eg DID document
|
|
|
|
# these are full feeds with multiple items, not just this one, so they're
|
|
# stored as audit records only. they're not used in to_as1. for Atom/RSS
|
|
# based Objects, our_as1 will be populated with an feed_index top-level
|
|
# integer field that indexes into one of these.
|
|
atom = ndb.TextProperty() # Atom XML
|
|
rss = ndb.TextProperty() # RSS XML
|
|
|
|
deleted = ndb.BooleanProperty()
|
|
|
|
# Copies of this object elsewhere, eg at:// URIs for ATProto records and
|
|
# nevent etc bech32-encoded Nostr ids, where this object is the original.
|
|
# Similar to u-syndication links in microformats2 and
|
|
# upstream/downstreamDuplicates in AS1.
|
|
copies = ndb.StructuredProperty(Target, repeated=True)
|
|
|
|
created = ndb.DateTimeProperty(auto_now_add=True)
|
|
updated = ndb.DateTimeProperty(auto_now=True)
|
|
|
|
new = None
|
|
changed = None
|
|
"""Protocol and subclasses set these in fetch if this :class:`Object` is
|
|
new or if its contents have changed from what was originally loaded from the
|
|
datastore. If either one is None, that means we don't know whether this
|
|
:class:`Object` is new/changed.
|
|
|
|
:attr:`changed` is populated by :meth:`activity_changed()`.
|
|
"""
|
|
|
|
lock = None
|
|
"""Initialized in __init__, synchronizes :meth:`add` and :meth:`remove`."""
|
|
|
|
# these were used for delivery tracking, but they were too expensive,
|
|
# so we stopped: https://github.com/snarfed/bridgy-fed/issues/1501
|
|
STATUSES = ('new', 'in progress', 'complete', 'failed', 'ignored')
|
|
status = ndb.StringProperty(choices=STATUSES)
|
|
delivered = ndb.StructuredProperty(Target, repeated=True)
|
|
undelivered = ndb.StructuredProperty(Target, repeated=True)
|
|
failed = ndb.StructuredProperty(Target, repeated=True)
|
|
|
|
@property
|
|
def as1(self):
|
|
def use_urls_as_ids(obj):
|
|
"""If id field is missing or not a URL, use the url field."""
|
|
id = obj.get('id')
|
|
if not id or not (util.is_web(id) or re.match(DOMAIN_RE, id)):
|
|
if url := util.get_url(obj):
|
|
obj['id'] = url
|
|
|
|
for field in 'author', 'actor', 'object':
|
|
if inner := as1.get_object(obj, field):
|
|
use_urls_as_ids(inner)
|
|
|
|
if self.our_as1:
|
|
obj = self.our_as1
|
|
if self.atom or self.rss:
|
|
use_urls_as_ids(obj)
|
|
|
|
elif self.as2:
|
|
obj = as2.to_as1(unwrap(self.as2))
|
|
|
|
elif self.bsky:
|
|
owner, _, _ = parse_at_uri(self.key.id())
|
|
ATProto = PROTOCOLS['atproto']
|
|
handle = ATProto(id=owner).handle
|
|
try:
|
|
obj = bluesky.to_as1(self.bsky, repo_did=owner, repo_handle=handle,
|
|
uri=self.key.id(), pds=ATProto.pds_for(self))
|
|
except (ValueError, RequestException):
|
|
logger.info(f"Couldn't convert to ATProto", exc_info=True)
|
|
return None
|
|
|
|
elif self.mf2:
|
|
obj = microformats2.json_to_object(self.mf2,
|
|
rel_urls=self.mf2.get('rel-urls'))
|
|
use_urls_as_ids(obj)
|
|
|
|
# use fetched final URL as id, not u-url
|
|
# https://github.com/snarfed/bridgy-fed/issues/829
|
|
if url := self.mf2.get('url'):
|
|
obj['id'] = (self.key.id() if self.key and '#' in self.key.id()
|
|
else url)
|
|
|
|
else:
|
|
return None
|
|
|
|
# populate id if necessary
|
|
if self.key:
|
|
obj.setdefault('id', self.key.id())
|
|
|
|
return obj
|
|
|
|
@ndb.ComputedProperty
|
|
def type(self): # AS1 objectType, or verb if it's an activity
|
|
if self.as1:
|
|
return as1.object_type(self.as1)
|
|
|
|
def __init__(self, *args, **kwargs):
|
|
super().__init__(*args, **kwargs)
|
|
self.lock = Lock()
|
|
|
|
def _expire(self):
|
|
"""Maybe automatically delete this Object after 90d using a TTL policy.
|
|
|
|
https://cloud.google.com/datastore/docs/ttl
|
|
|
|
They recommend not indexing TTL properties:
|
|
https://cloud.google.com/datastore/docs/ttl#ttl_properties_and_indexes
|
|
"""
|
|
if self.type not in DONT_EXPIRE_OBJECT_TYPES:
|
|
return (self.updated or util.now()) + OBJECT_EXPIRE_AGE
|
|
|
|
expire = ndb.ComputedProperty(_expire, indexed=False)
|
|
|
|
def _pre_put_hook(self):
|
|
"""
|
|
* Validate that at:// URIs have DID repos
|
|
* Set/remove the activity label
|
|
* Strip @context from as2 (we don't do LD) to save disk space
|
|
"""
|
|
id = self.key.id()
|
|
|
|
if self.source_protocol not in (None, 'ui'):
|
|
proto = PROTOCOLS[self.source_protocol]
|
|
assert proto.owns_id(id) is not False, \
|
|
f'Protocol {proto.LABEL} does not own id {id}'
|
|
|
|
if id.startswith('at://'):
|
|
repo, _, _ = parse_at_uri(id)
|
|
if not repo.startswith('did:'):
|
|
# TODO: if we hit this, that means the AppView gave us an AT URI
|
|
# with a handle repo/authority instead of DID. that's surprising!
|
|
# ...if so, and if we need to handle it, add a new
|
|
# arroba.did.canonicalize_at_uri() function, then use it here,
|
|
# or before.
|
|
raise ValueError(
|
|
f'at:// URI ids must have DID repos; got {id}')
|
|
|
|
if self.as1 and self.as1.get('objectType') == 'activity':
|
|
self.add('labels', 'activity')
|
|
elif 'activity' in self.labels:
|
|
self.remove('labels', 'activity')
|
|
|
|
if self.as2:
|
|
self.as2.pop('@context', None)
|
|
for field in 'actor', 'attributedTo', 'author', 'object':
|
|
for val in util.get_list(self.as2, field):
|
|
if isinstance(val, dict):
|
|
val.pop('@context', None)
|
|
|
|
def _post_put_hook(self, future):
|
|
# TODO: assert that as1 id is same as key id? in pre put hook?
|
|
logger.debug(f'Wrote {self.key}')
|
|
|
|
@classmethod
|
|
def get_by_id(cls, id, authed_as=None, **kwargs):
|
|
"""Fetches the :class:`Object` with the given id, if it exists.
|
|
|
|
Args:
|
|
id (str)
|
|
authed_as (str): optional; if provided, and a matching :class:`Object`
|
|
already exists, its ``author`` or ``actor`` must contain this actor
|
|
id. Implements basic authorization for updates and deletes.
|
|
|
|
Returns:
|
|
Object:
|
|
|
|
Raises:
|
|
:class:`werkzeug.exceptions.Forbidden` if ``authed_as`` doesn't match
|
|
the existing object
|
|
"""
|
|
obj = super().get_by_id(id, **kwargs)
|
|
|
|
if obj and obj.as1 and authed_as:
|
|
# authorization: check that the authed user is allowed to modify
|
|
# this object
|
|
# https://www.w3.org/wiki/ActivityPub/Primer/Authentication_Authorization
|
|
proto = PROTOCOLS.get(obj.source_protocol)
|
|
assert proto, obj.source_protocol
|
|
owners = [ids.normalize_user_id(id=owner, proto=proto)
|
|
for owner in (as1.get_ids(obj.as1, 'author')
|
|
+ as1.get_ids(obj.as1, 'actor'))
|
|
+ [id]]
|
|
if (ids.normalize_user_id(id=authed_as, proto=proto) not in owners
|
|
and ids.profile_id(id=authed_as, proto=proto) not in owners):
|
|
report_error("Auth: Object: authed_as doesn't match owner",
|
|
user=f'{id} authed_as {authed_as} owners {owners}')
|
|
error(f"authed user {authed_as} isn't object owner {owners}",
|
|
status=403)
|
|
|
|
return obj
|
|
|
|
@classmethod
|
|
def get_or_create(cls, id, authed_as=None, **props):
|
|
"""Returns an :class:`Object` with the given property values.
|
|
|
|
If a matching :class:`Object` doesn't exist in the datastore, creates it
|
|
first. Only populates non-False/empty property values in props into the
|
|
object. Also populates the :attr:`new` and :attr:`changed` properties.
|
|
|
|
Not transactional because transactions don't read or write memcache. :/
|
|
Fortunately we don't really depend on atomicity for anything, last
|
|
writer wins is pretty much always fine.
|
|
|
|
Args:
|
|
authed_as (str): optional; if provided, and a matching :class:`Object`
|
|
already exists, its ``author`` or ``actor`` must contain this actor
|
|
id. Implements basic authorization for updates and deletes.
|
|
|
|
Returns:
|
|
Object:
|
|
|
|
Raises:
|
|
:class:`werkzeug.exceptions.Forbidden` if ``authed_as`` doesn't match
|
|
the existing object
|
|
"""
|
|
obj = cls.get_by_id(id, authed_as=authed_as)
|
|
if obj:
|
|
obj.new = False
|
|
if orig_as1 := obj.as1:
|
|
assert authed_as
|
|
else:
|
|
obj = Object(id=id)
|
|
obj.new = True
|
|
|
|
obj.changed = None
|
|
for field in 'new', 'changed':
|
|
val = props.pop(field, None)
|
|
if val is not None:
|
|
setattr(obj, field, val)
|
|
|
|
if set(props.keys()) & set(('as2', 'bsky', 'mf2', 'raw')):
|
|
obj.clear()
|
|
|
|
# merge repeated fields
|
|
for field in 'feed', 'copies', 'labels', 'notify', 'users':
|
|
getattr(obj, field).extend(props.pop(field, []))
|
|
|
|
obj.populate(**{
|
|
k: v for k, v in props.items()
|
|
if v and not isinstance(getattr(Object, k), ndb.ComputedProperty)
|
|
})
|
|
if not obj.new and obj.changed is None:
|
|
obj.changed = obj.activity_changed(orig_as1)
|
|
|
|
obj.put()
|
|
return obj
|
|
|
|
def add(self, prop, val):
|
|
"""Adds a value to a multiply-valued property. Uses ``self.lock``.
|
|
|
|
Args:
|
|
prop (str)
|
|
val
|
|
"""
|
|
with self.lock:
|
|
added = util.add(getattr(self, prop), val)
|
|
|
|
if prop == 'copies' and added:
|
|
memcache.pickle_memcache.set(memcache.memoize_key(
|
|
get_original_object_key, val.uri), self.key)
|
|
|
|
def remove(self, prop, val):
|
|
"""Removes a value from a multiply-valued property. Uses ``self.lock``.
|
|
|
|
Args:
|
|
prop (str)
|
|
val
|
|
"""
|
|
with self.lock:
|
|
getattr(self, prop).remove(val)
|
|
|
|
def clear(self):
|
|
"""Clears the :attr:`Object.our_as1` property."""
|
|
self.our_as1 = None
|
|
|
|
@staticmethod
|
|
def from_request():
|
|
"""Creates and returns an :class:`Object` from form-encoded JSON parameters.
|
|
|
|
Parameters:
|
|
obj_id (str): id of :class:`models.Object` to handle
|
|
*: If ``obj_id`` is unset, all other parameters are properties for a
|
|
new :class:`models.Object` to handle
|
|
"""
|
|
if obj_id := request.form.get('obj_id'):
|
|
return Object.get_by_id(obj_id)
|
|
|
|
props = {field: request.form.get(field)
|
|
for field in ('id', 'atom', 'rss', 'source_protocol')}
|
|
|
|
for json_prop in 'as2', 'bsky', 'mf2', 'our_as1', 'raw':
|
|
if val := request.form.get(json_prop):
|
|
props[json_prop] = json_loads(val)
|
|
|
|
obj = Object(**props)
|
|
if not obj.key and obj.as1:
|
|
if id := obj.as1.get('id'):
|
|
obj.key = ndb.Key(Object, id)
|
|
|
|
return obj
|
|
|
|
def to_request(self):
|
|
"""Returns a query parameter dict representing this :class:`Object`."""
|
|
form = {}
|
|
|
|
for json_prop in 'as2', 'bsky', 'mf2', 'our_as1', 'raw':
|
|
if val := getattr(self, json_prop, None):
|
|
form[json_prop] = json_dumps(val, sort_keys=True)
|
|
|
|
for prop in 'atom', 'rss', 'source_protocol':
|
|
if val := getattr(self, prop):
|
|
form[prop] = val
|
|
|
|
if self.key:
|
|
form['id'] = self.key.id()
|
|
|
|
return form
|
|
|
|
def activity_changed(self, other_as1):
|
|
"""Returns True if this activity is meaningfully changed from ``other_as1``.
|
|
|
|
...otherwise False.
|
|
|
|
Used to populate :attr:`changed`.
|
|
|
|
Args:
|
|
other_as1 (dict): AS1 object, or none
|
|
"""
|
|
# ignore inReplyTo since we translate it between protocols
|
|
return (as1.activity_changed(self.as1, other_as1, inReplyTo=False)
|
|
if self.as1 and other_as1
|
|
else bool(self.as1) != bool(other_as1))
|
|
|
|
def actor_link(self, image=True, sized=False, user=None):
|
|
"""Returns a pretty HTML link with the actor's name and picture.
|
|
|
|
TODO: unify with :meth:`User.user_link`?
|
|
|
|
Args:
|
|
image (bool): whether to include an ``img`` tag with the actor's picture
|
|
sized (bool): whether to set an explicit (``width=32``) size on the
|
|
profile picture ``img` tag
|
|
user (User): current user
|
|
|
|
Returns:
|
|
str:
|
|
"""
|
|
attrs = {'class': 'h-card u-author'}
|
|
|
|
if user and (user.key in self.users or user.key.id() in self.domains):
|
|
# outbound; show a nice link to the user
|
|
return user.user_link(handle=False, pictures=True)
|
|
|
|
proto = PROTOCOLS.get(self.source_protocol)
|
|
|
|
actor = None
|
|
if self.as1:
|
|
actor = (as1.get_object(self.as1, 'actor')
|
|
or as1.get_object(self.as1, 'author'))
|
|
# hydrate from datastore if available
|
|
# TODO: optimize! this is called serially in loops, eg in home.html
|
|
if set(actor.keys()) == {'id'} and self.source_protocol:
|
|
actor_obj = proto.load(actor['id'], remote=False)
|
|
if actor_obj and actor_obj.as1:
|
|
actor = actor_obj.as1
|
|
|
|
if not actor:
|
|
return ''
|
|
elif set(actor.keys()) == {'id'}:
|
|
return common.pretty_link(actor['id'], attrs=attrs, user=user)
|
|
|
|
url = as1.get_url(actor)
|
|
name = actor.get('displayName') or actor.get('username') or ''
|
|
img_url = util.get_url(actor, 'image')
|
|
if not image or not img_url:
|
|
return common.pretty_link(url, text=name, attrs=attrs, user=user)
|
|
|
|
logo = ''
|
|
if proto:
|
|
logo = f'<span class="logo" title="{self.__class__.__name__}">{proto.LOGO_HTML}</span>'
|
|
|
|
return f"""\
|
|
{logo}
|
|
<a class="h-card u-author" href="{url}" title="{name}">
|
|
<img class="profile" src="{img_url}" {'width="32"' if sized else ''}/>
|
|
<span style="unicode-bidi: isolate">{util.ellipsize(name, chars=40)}</span>
|
|
</a>"""
|
|
|
|
def get_copy(self, proto):
|
|
"""Returns the id for the copy of this object in a given protocol.
|
|
|
|
...or None if no such copy exists. If ``proto`` is ``source_protocol``,
|
|
returns this object's key id.
|
|
|
|
Args:
|
|
proto: :class:`Protocol` subclass
|
|
|
|
Returns:
|
|
str:
|
|
"""
|
|
if self.source_protocol in (proto.LABEL, proto.ABBREV):
|
|
return self.key.id()
|
|
|
|
for copy in self.copies:
|
|
if copy.protocol in (proto.LABEL, proto.ABBREV):
|
|
return copy.uri
|
|
|
|
def resolve_ids(self):
|
|
"""Resolves "copy" ids, subdomain ids, etc with their originals.
|
|
|
|
The end result is that all ids are original "source" ids, ie in the
|
|
protocol that they first came from.
|
|
|
|
Specifically, resolves:
|
|
|
|
* ids in :class:`User.copies` and :class:`Object.copies`, eg ATProto
|
|
records and Nostr events that we bridged, to the ids of their
|
|
original objects in their source protocol, eg
|
|
``at://did:plc:abc/app.bsky.feed.post/123`` => ``https://mas.to/@user/456``.
|
|
* Bridgy Fed subdomain URLs to the ids embedded inside them, eg
|
|
``https://bsky.brid.gy/ap/did:plc:xyz`` => ``did:plc:xyz``
|
|
* ATProto bsky.app URLs to their DIDs or `at://` URIs, eg
|
|
``https://bsky.app/profile/a.com`` => ``did:plc:123``
|
|
|
|
...in these AS1 fields, in place:
|
|
|
|
* ``id``
|
|
* ``actor``
|
|
* ``author``
|
|
* ``object``
|
|
* ``object.actor``
|
|
* ``object.author``
|
|
* ``object.id``
|
|
* ``object.inReplyTo``
|
|
* ``tags.[objectType=mention].url``
|
|
|
|
:meth:`protocol.Protocol.translate_ids` is partly the inverse of this.
|
|
Much of the same logic is duplicated there!
|
|
|
|
TODO: unify with :meth:`normalize_ids`, :meth:`Object.normalize_ids`.
|
|
"""
|
|
if not self.as1:
|
|
return
|
|
|
|
# extract ids, strip Bridgy Fed subdomain URLs
|
|
outer_obj = unwrap(self.as1)
|
|
if outer_obj != self.as1:
|
|
self.our_as1 = util.trim_nulls(outer_obj)
|
|
|
|
self_proto = PROTOCOLS.get(self.source_protocol)
|
|
if not self_proto:
|
|
return
|
|
|
|
inner_obj = outer_obj['object'] = as1.get_object(outer_obj)
|
|
replaced = False
|
|
|
|
def replace(val, orig_fn):
|
|
id = val.get('id') if isinstance(val, dict) else val
|
|
if not id:
|
|
return id
|
|
|
|
orig = orig_fn(id)
|
|
if not orig:
|
|
return val
|
|
|
|
nonlocal replaced
|
|
replaced = True
|
|
logger.debug(f'Resolved copy id {val} to original {orig.id()}')
|
|
|
|
if isinstance(val, dict) and util.trim_nulls(val).keys() > {'id'}:
|
|
val['id'] = orig.id()
|
|
return val
|
|
else:
|
|
return orig.id()
|
|
|
|
# actually replace ids
|
|
#
|
|
# object field could be either object (eg repost) or actor (eg follow)
|
|
outer_obj['object'] = replace(inner_obj, get_original_object_key)
|
|
if not replaced:
|
|
outer_obj['object'] = replace(inner_obj, get_original_user_key)
|
|
|
|
for obj in outer_obj, inner_obj:
|
|
for tag in as1.get_objects(obj, 'tags'):
|
|
if tag.get('objectType') == 'mention':
|
|
tag['url'] = replace(tag.get('url'), get_original_user_key)
|
|
for field, fn in (
|
|
('actor', get_original_user_key),
|
|
('author', get_original_user_key),
|
|
('inReplyTo', get_original_object_key),
|
|
):
|
|
obj[field] = [replace(val, fn) for val in util.get_list(obj, field)]
|
|
if len(obj[field]) == 1:
|
|
obj[field] = obj[field][0]
|
|
|
|
if replaced:
|
|
self.our_as1 = util.trim_nulls(outer_obj)
|
|
|
|
def normalize_ids(self):
|
|
"""Normalizes ids to their protocol's canonical representation, if any.
|
|
|
|
For example, normalizes ATProto ``https://bsky.app/...`` URLs to DIDs
|
|
for profiles, ``at://`` URIs for posts.
|
|
|
|
Modifies this object in place.
|
|
|
|
TODO: unify with :meth:`resolve_ids`, :meth:`Protocol.translate_ids`.
|
|
"""
|
|
from protocol import Protocol
|
|
|
|
if not self.as1:
|
|
return
|
|
|
|
logger.debug(f'Normalizing ids')
|
|
outer_obj = copy.deepcopy(self.as1)
|
|
inner_objs = as1.get_objects(outer_obj)
|
|
replaced = False
|
|
|
|
def replace(val, translate_fn):
|
|
nonlocal replaced
|
|
|
|
orig = val.get('id') if isinstance(val, dict) else val
|
|
if not orig:
|
|
return val
|
|
|
|
proto = Protocol.for_id(orig, remote=False)
|
|
if not proto:
|
|
return val
|
|
|
|
translated = translate_fn(id=orig, from_=proto, to=proto)
|
|
if translated and translated != orig:
|
|
# logger.debug(f'Normalized {proto.LABEL} id {orig} to {translated}')
|
|
replaced = True
|
|
if isinstance(val, dict):
|
|
val['id'] = translated
|
|
return val
|
|
else:
|
|
return translated
|
|
|
|
return val
|
|
|
|
# actually replace ids
|
|
for obj in [outer_obj] + inner_objs:
|
|
for tag in as1.get_objects(obj, 'tags'):
|
|
if tag.get('objectType') == 'mention':
|
|
tag['url'] = replace(tag.get('url'), ids.translate_user_id)
|
|
for field in ['actor', 'author', 'inReplyTo']:
|
|
fn = (ids.translate_object_id if field == 'inReplyTo'
|
|
else ids.translate_user_id)
|
|
obj[field] = [replace(val, fn) for val in util.get_list(obj, field)]
|
|
if len(obj[field]) == 1:
|
|
obj[field] = obj[field][0]
|
|
|
|
outer_obj['object'] = []
|
|
for inner_obj in inner_objs:
|
|
translate_fn = (ids.translate_user_id
|
|
if (as1.object_type(inner_obj) in as1.ACTOR_TYPES
|
|
or as1.object_type(outer_obj) in
|
|
('follow', 'stop-following'))
|
|
else ids.translate_object_id)
|
|
|
|
got = replace(inner_obj, translate_fn)
|
|
if isinstance(got, dict) and util.trim_nulls(got).keys() == {'id'}:
|
|
got = got['id']
|
|
|
|
outer_obj['object'].append(got)
|
|
|
|
if len(outer_obj['object']) == 1:
|
|
outer_obj['object'] = outer_obj['object'][0]
|
|
|
|
if replaced:
|
|
self.our_as1 = util.trim_nulls(outer_obj)
|
|
|
|
|
|
class Follower(ndb.Model):
|
|
"""A follower of a Bridgy Fed user."""
|
|
STATUSES = ('active', 'inactive')
|
|
|
|
# these are both subclasses of User
|
|
from_ = ndb.KeyProperty(name='from', required=True)
|
|
to = ndb.KeyProperty(required=True)
|
|
|
|
follow = ndb.KeyProperty(Object) # last follow activity
|
|
status = ndb.StringProperty(choices=STATUSES, default='active')
|
|
|
|
created = ndb.DateTimeProperty(auto_now_add=True)
|
|
updated = ndb.DateTimeProperty(auto_now=True)
|
|
|
|
# OLD. some stored entities still have these; do not reuse.
|
|
# src = ndb.StringProperty()
|
|
# dest = ndb.StringProperty()
|
|
# last_follow = JsonProperty()
|
|
|
|
def _pre_put_hook(self):
|
|
# we're a bridge! stick with bridging.
|
|
assert self.from_.kind() != self.to.kind(), f'from {self.from_} to {self.to}'
|
|
|
|
def _post_put_hook(self, future):
|
|
logger.debug(f'Wrote {self.key}')
|
|
|
|
@classmethod
|
|
def get_or_create(cls, *, from_, to, **kwargs):
|
|
"""Returns a Follower with the given ``from_`` and ``to`` users.
|
|
|
|
Not transactional because transactions don't read or write memcache. :/
|
|
Fortunately we don't really depend on atomicity for anything, last
|
|
writer wins is pretty much always fine.
|
|
|
|
If a matching :class:`Follower` doesn't exist in the datastore, creates
|
|
it first.
|
|
|
|
Args:
|
|
from_ (User)
|
|
to (User)
|
|
|
|
Returns:
|
|
Follower:
|
|
"""
|
|
assert from_
|
|
assert to
|
|
|
|
follower = Follower.query(Follower.from_ == from_.key,
|
|
Follower.to == to.key,
|
|
).get()
|
|
if not follower:
|
|
follower = Follower(from_=from_.key, to=to.key, **kwargs)
|
|
follower.put()
|
|
elif kwargs:
|
|
# update existing entity with new property values, eg to make an
|
|
# inactive Follower active again
|
|
for prop, val in kwargs.items():
|
|
setattr(follower, prop, val)
|
|
follower.put()
|
|
|
|
return follower
|
|
|
|
@staticmethod
|
|
def fetch_page(collection, user):
|
|
r"""Fetches a page of :class:`Follower`\s for a given user.
|
|
|
|
Wraps :func:`fetch_page`. Paging uses the ``before`` and ``after`` query
|
|
parameters, if available in the request.
|
|
|
|
Args:
|
|
collection (str): ``followers`` or ``following``
|
|
user (User)
|
|
|
|
Returns:
|
|
(list of Follower, str, str) tuple: results, annotated with an extra
|
|
``user`` attribute that holds the follower or following :class:`User`,
|
|
and new str query param values for ``before`` and ``after`` to fetch
|
|
the previous and next pages, respectively
|
|
"""
|
|
assert collection in ('followers', 'following'), collection
|
|
|
|
filter_prop = Follower.to if collection == 'followers' else Follower.from_
|
|
query = Follower.query(
|
|
Follower.status == 'active',
|
|
filter_prop == user.key,
|
|
)
|
|
|
|
followers, before, after = fetch_page(query, Follower, by=Follower.updated)
|
|
users = ndb.get_multi(f.from_ if collection == 'followers' else f.to
|
|
for f in followers)
|
|
User.load_multi(u for u in users if u)
|
|
|
|
for f, u in zip(followers, users):
|
|
f.user = u
|
|
followers = [f for f in followers if not f.user.status]
|
|
|
|
return followers, before, after
|
|
|
|
|
|
def fetch_objects(query, by=None, user=None):
|
|
"""Fetches a page of :class:`Object` entities from a datastore query.
|
|
|
|
Wraps :func:`fetch_page` and adds attributes to the returned
|
|
:class:`Object` entities for rendering in ``objects.html``.
|
|
|
|
Args:
|
|
query (ndb.Query)
|
|
by (ndb.model.Property): either :attr:`Object.updated` or
|
|
:attr:`Object.created`
|
|
user (User): current user
|
|
|
|
Returns:
|
|
(list of Object, str, str) tuple:
|
|
(results, new ``before`` query param, new ``after`` query param)
|
|
to fetch the previous and next pages, respectively
|
|
"""
|
|
assert by is Object.updated or by is Object.created
|
|
objects, new_before, new_after = fetch_page(query, Object, by=by)
|
|
objects = [o for o in objects if as1.is_public(o.as1) and not o.deleted]
|
|
|
|
# synthesize human-friendly content for objects
|
|
for i, obj in enumerate(objects):
|
|
obj_as1 = obj.as1
|
|
inner_obj = as1.get_object(obj_as1)
|
|
|
|
# synthesize text snippet
|
|
type = as1.object_type(obj_as1)
|
|
if type == 'post':
|
|
inner_type = inner_obj.get('objectType')
|
|
if inner_type:
|
|
type = inner_type
|
|
|
|
# AS1 verb => human-readable phrase
|
|
phrases = {
|
|
'accept': 'accepted',
|
|
'article': 'posted',
|
|
'comment': 'replied',
|
|
'delete': 'deleted',
|
|
'follow': 'followed',
|
|
'invite': 'is invited to',
|
|
'issue': 'filed issue',
|
|
'like': 'liked',
|
|
'note': 'posted',
|
|
'post': 'posted',
|
|
'repost': 'reposted',
|
|
'rsvp-interested': 'is interested in',
|
|
'rsvp-maybe': 'might attend',
|
|
'rsvp-no': 'is not attending',
|
|
'rsvp-yes': 'is attending',
|
|
'share': 'reposted',
|
|
'stop-following': 'unfollowed',
|
|
'undo': 'undid',
|
|
'update': 'updated',
|
|
}
|
|
obj.phrase = phrases.get(type)
|
|
|
|
content = (inner_obj.get('content')
|
|
or inner_obj.get('displayName')
|
|
or inner_obj.get('summary'))
|
|
if content:
|
|
content = util.parse_html(content).get_text()
|
|
|
|
urls = as1.object_urls(inner_obj)
|
|
id = unwrap(inner_obj.get('id', ''))
|
|
url = urls[0] if urls else id
|
|
if (type == 'update' and
|
|
(obj.users and (user.is_web_url(id)
|
|
or id.strip('/') == obj.users[0].id())
|
|
or obj.domains and id.strip('/') == f'https://{obj.domains[0]}')):
|
|
obj.phrase = 'updated'
|
|
obj_as1.update({
|
|
'content': 'their profile',
|
|
'url': id,
|
|
})
|
|
elif url and not content:
|
|
# heuristics for sniffing URLs and converting them to more friendly
|
|
# phrases and user handles.
|
|
# TODO: standardize this into granary.as2 somewhere?
|
|
from activitypub import FEDI_URL_RE
|
|
from atproto import COLLECTION_TO_TYPE, did_to_handle
|
|
|
|
handle = suffix = ''
|
|
if match := FEDI_URL_RE.match(url):
|
|
handle = match.group(2)
|
|
if match.group(4):
|
|
suffix = "'s post"
|
|
elif match := BSKY_APP_URL_RE.match(url):
|
|
handle = match.group('id')
|
|
if match.group('tid'):
|
|
suffix = "'s post"
|
|
elif match := AT_URI_PATTERN.match(url):
|
|
handle = match.group('repo')
|
|
if coll := match.group('collection'):
|
|
suffix = f"'s {COLLECTION_TO_TYPE.get(coll) or 'post'}"
|
|
url = bluesky.at_uri_to_web_url(url)
|
|
elif url.startswith('did:'):
|
|
handle = url
|
|
url = bluesky.Bluesky.user_url(handle)
|
|
|
|
if handle:
|
|
if handle.startswith('did:'):
|
|
handle = did_to_handle(handle) or handle
|
|
content = f'@{handle}{suffix}'
|
|
|
|
if url:
|
|
content = common.pretty_link(url, text=content, user=user)
|
|
|
|
obj.content = (obj_as1.get('content')
|
|
or obj_as1.get('displayName')
|
|
or obj_as1.get('summary'))
|
|
obj.url = util.get_first(obj_as1, 'url')
|
|
|
|
if type in ('like', 'follow', 'repost', 'share') or not obj.content:
|
|
if obj.url:
|
|
obj.phrase = common.pretty_link(
|
|
obj.url, text=obj.phrase, attrs={'class': 'u-url'}, user=user)
|
|
if content:
|
|
obj.content = content
|
|
obj.url = url
|
|
|
|
return objects, new_before, new_after
|
|
|
|
|
|
def fetch_page(query, model_class, by=None):
|
|
"""Fetches a page of results from a datastore query.
|
|
|
|
Uses the ``before`` and ``after`` query params (if provided; should be
|
|
ISO8601 timestamps) and the ``by`` property to identify the page to fetch.
|
|
|
|
Populates a ``log_url_path`` property on each result entity that points to a
|
|
its most recent logged request.
|
|
|
|
Args:
|
|
query (google.cloud.ndb.query.Query)
|
|
model_class (class)
|
|
by (ndb.model.Property): paging property, eg :attr:`Object.updated`
|
|
or :attr:`Object.created`
|
|
|
|
Returns:
|
|
(list of Object or Follower, str, str) tuple: (results, new_before,
|
|
new_after), where new_before and new_after are query param values for
|
|
``before`` and ``after`` to fetch the previous and next pages,
|
|
respectively
|
|
"""
|
|
assert by
|
|
|
|
# if there's a paging param ('before' or 'after'), update query with it
|
|
# TODO: unify this with Bridgy's user page
|
|
def get_paging_param(param):
|
|
val = request.values.get(param)
|
|
if val:
|
|
try:
|
|
dt = util.parse_iso8601(val.replace(' ', '+'))
|
|
except BaseException as e:
|
|
error(f"Couldn't parse {param}, {val!r} as ISO8601: {e}")
|
|
if dt.tzinfo:
|
|
dt = dt.astimezone(timezone.utc).replace(tzinfo=None)
|
|
return dt
|
|
|
|
before = get_paging_param('before')
|
|
after = get_paging_param('after')
|
|
if before and after:
|
|
error("can't handle both before and after")
|
|
elif after:
|
|
query = query.filter(by >= after).order(by)
|
|
elif before:
|
|
query = query.filter(by < before).order(-by)
|
|
else:
|
|
query = query.order(-by)
|
|
|
|
query_iter = query.iter()
|
|
results = sorted(itertools.islice(query_iter, 0, PAGE_SIZE),
|
|
key=lambda r: r.updated, reverse=True)
|
|
|
|
# calculate new paging param(s)
|
|
has_next = results and query_iter.probably_has_next()
|
|
new_after = (
|
|
before if before
|
|
else results[0].updated if has_next and after
|
|
else None)
|
|
if new_after:
|
|
new_after = new_after.isoformat()
|
|
|
|
new_before = (
|
|
after if after else
|
|
results[-1].updated if has_next
|
|
else None)
|
|
if new_before:
|
|
new_before = new_before.isoformat()
|
|
|
|
return results, new_before, new_after
|
|
|
|
|
|
@lru_cache(maxsize=100000)
|
|
@memcache.memoize(expire=GET_ORIGINALS_CACHE_EXPIRATION)
|
|
def get_original_object_key(copy_id):
|
|
"""Finds the :class:`Object` with a given copy id, if any.
|
|
|
|
Note that :meth:`Object.add` also updates this function's
|
|
:func:`memcache.memoize` cache.
|
|
|
|
Args:
|
|
copy_id (str)
|
|
|
|
Returns:
|
|
google.cloud.ndb.Key or None
|
|
"""
|
|
assert copy_id
|
|
|
|
return Object.query(Object.copies.uri == copy_id).get(keys_only=True)
|
|
|
|
|
|
@lru_cache(maxsize=100000)
|
|
@memcache.memoize(expire=GET_ORIGINALS_CACHE_EXPIRATION)
|
|
def get_original_user_key(copy_id):
|
|
"""Finds the user with a given copy id, if any.
|
|
|
|
Note that :meth:`User.add` also updates this function's
|
|
:func:`memcache.memoize` cache.
|
|
|
|
Args:
|
|
copy_id (str)
|
|
not_proto (Protocol): optional, don't query this protocol
|
|
|
|
Returns:
|
|
google.cloud.ndb.Key or None
|
|
"""
|
|
assert copy_id
|
|
|
|
for proto in PROTOCOLS.values():
|
|
if proto and proto.LABEL != 'ui' and not proto.owns_id(copy_id):
|
|
if orig := proto.query(proto.copies.uri == copy_id).get(keys_only=True):
|
|
return orig
|