Something went wrong. Try again.
🌉 A bridge between decentralized social networks fed.brid.gy
Something went wrong. Try again.
bridgy-fed common.py
14 kB · 426 lines
Python
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427"""Misc common utilities."""from datetime import timedeltaimport functoolsimport loggingimport osfrom pathlib import Pathimport reimport stringimport threadingimport urllib.parsefrom urllib.parse import urljoin, urlparse
import flaskfrom flask import abort, g, has_request_context, make_response, redirect, requestfrom flask.views import Viewfrom google.api_core.exceptions import InvalidArgumentfrom google.cloud.error_reporting.util import build_flask_contextfrom google.cloud import ndbfrom google.cloud.ndb.key import Keyfrom google.protobuf.timestamp_pb2 import Timestampfrom granary import as2import jwtfrom negotiator import ContentNegotiator, AcceptParameters, ContentTypeimport requestsfrom webutil import flask_util, models, util, webmentionfrom webutil.appengine_config import error_reporting_client, tasks_clientfrom webutil.appengine_info import DEBUG, LOCAL_SERVER, READ_ONLYfrom webutil.util import interpret_http_exception, json_dumpsimport werkzeug.exceptionsfrom werkzeug.exceptions import HTTPException
import configfrom domains import ( DOMAIN_RE, DOMAINS, LOCAL_DOMAINS, OTHER_DOMAINS, PROTOCOL_DOMAINS, PRIMARY_DOMAIN, SUBDOMAIN_BASE_URL_RE, SUPERDOMAIN,)import memcache
logger = logging.getLogger(__name__)
CONTENT_TYPE_HTML = 'text/html; charset=utf-8'
GCP_PROJECT_ID = 'bridgy-federated' # used in create_task
CACHE_CONTROL = {'Cache-Control': 'public, max-age=3600'} # 1 hourCACHE_CONTROL_VARY_ACCEPT = {**CACHE_CONTROL, 'Vary': 'Accept'}
USER_AGENT = 'Bridgy Fed (https://fed.brid.gy/)'util.set_user_agent(USER_AGENT)
# the GCP load balancer collapses :// down to :/ anywhere in a URL path, so we# expand it back, for ids, for URIs embedded in query params, and for full pathsCOLLAPSED_SCHEME_RE = re.compile(r'([a-z][a-z0-9+.-]*:/)(?=[^/])', re.IGNORECASE)
# https://cloud.google.com/appengine/docs/locationsTASKS_LOCATION = 'us-central1'RUN_TASKS_INLINE = False # overridden by unit tests
# for Protocol.REQUIRES_OLD_ACCOUNT, how old is old enoughOLD_ACCOUNT_AGE = timedelta(days=7)
# populated later in this fileNDB_CONTEXT_KWARGS = None
_negotiator = ContentNegotiator(acceptable=[ AcceptParameters(ContentType(CONTENT_TYPE_HTML)), AcceptParameters(ContentType(as2.CONTENT_TYPE)), AcceptParameters(ContentType(as2.CONTENT_TYPE_LD)),])
# User ids who opt into testing new "beta" features and changes before we roll them# out to everyone.with open(Path(os.path.dirname(__file__)) / 'beta_users.txt') as f: BETA_USER_IDS = util.load_file_lines(f)
class ErrorButDoNotRetryTask(HTTPException): code = 299 description = 'ErrorButDoNotRetryTask'
# https://github.com/pallets/flask/issues/1837#issuecomment-304996942werkzeug.exceptions.default_exceptions.setdefault(299, ErrorButDoNotRetryTask)werkzeug.exceptions._aborter.mapping.setdefault(299, ErrorButDoNotRetryTask)
class classproperty: """@classmethod + @property.
Python 3.13 disabled combining classmethod with other decorators like property, so we use this now.
https://docs.python.org/3.13/whatsnew/3.13.html#builtins """ def __init__(self, method): self.method = method
def __get__(self, obj, cls=None): if cls is None: cls = type(obj) return self.method(cls)
@functools.cachedef bot_user_ids(): """Returns all copy ids for protocol bot users.""" from models import PROTOCOLS
bot_ids = set(PROTOCOL_DOMAINS) protocols = set(p for p in PROTOCOLS.values() if p and p.bot_user_id())
for bot_proto in protocols: if not (bot := bot_proto.bot_user()): continue
bot_ids.update(copy.uri for copy in bot.copies)
for other_proto in protocols: if (bot_proto != other_proto and not other_proto.HAS_COPIES and other_proto.LABEL not in bot_proto.DEFAULT_ENABLED_PROTOCOLS): bot_ids.add(bot.id_as(other_proto))
return bot_ids
def error(err, status=400, exc_info=None, **kwargs): """Like :func:`webutil.flask_util.error`, but wraps body in JSON.""" msg = str(err) logger.info(f'Returning {status}: {msg}', exc_info=exc_info) abort(status, response=make_response({'error': msg}, status), **kwargs)
def pretty_link(url, text=None, user=None, **kwargs): """Wrapper around :func:`webutil.util.pretty_link` that converts Mastodon user URLs to @-@ handles.
Eg for URLs like https://mastodon.social/@foo and https://mastodon.social/users/foo, defaults text to ``@foo@mastodon.social`` if it's not provided.
Args: url (str) text (str) user (models.User): current user kwargs: passed through to :func:`webutil.util.pretty_link` """ if user and user.is_web_url(url): return user.html_link(handle=False, pictures=True)
if text is None: match = re.match(r'https?://([^/]+)/(@|users/)([^/]+)$', url) if match: text = match.expand(r'@\3@\1')
return util.pretty_link(url, text=text, **kwargs)
def content_type(resp): """Returns a :class:`requests.Response`'s Content-Type, without charset suffix.""" type = resp.headers.get('Content-Type') if type: # TODO: don't remove profile # right now, when we remove it, and don't use it to compare against eg # as2.CONTENT_TYPE_LD, we end up accepting non-AS2 JSON-LD, eg: # Content-Type: application/ld+json; charset=UTF-8 return type.split(';')[0]
def create_task(queue, app_id=GCP_PROJECT_ID, delay=None, app=None, **params): """Adds a Cloud Tasks task.
If running in a local server, runs the task handler inline instead of creating a task.
Args: queue (str): queue name delay (:class:`datetime.timedelta`): optional, used as task ETA (from now) app (flask.Flask): if not provided, defaults to ``router.app`` params: form-encoded and included in the task request body
Returns: flask.Response or (str, int): response from either running the task inline, if running in a local server, or the response from creating the task. """ assert queue path = f'/queue/{queue}'
params = { k: json_dumps(v, sort_keys=True) if isinstance(v, dict) else v for k, v in params.items() if v is not None }
try: traceparent = request.headers.get('traceparent') or '' cloud_trace = request.headers.get('X-Cloud-Trace-Context') or '' except RuntimeError: # not currently in a request context traceparent = cloud_trace = ''
if RUN_TASKS_INLINE or LOCAL_SERVER: logger.info(f'Running task inline: {queue} {params}') if not app: from router import app return app.test_client().post(path, data=params, headers={ flask_util.CLOUD_TASKS_TASK_HEADER: 'inline', })
# # alternative: run inline in this request context # request.form = params # endpoint, args = app.url_map.bind(request.server[0])\ # .match(path, method='POST') # return app.view_functions[endpoint](**args)
# determine task ETA eta = None now = util.now() if authed_as := params.get('authed_as'): eta = memcache.task_eta(queue, authed_as, protocol=params.get('protocol'))
if delay: if not eta: eta = now eta += delay
schedule_time = None delay_msg = 'now' if eta and eta > now + timedelta(seconds=1): schedule_time = Timestamp(seconds=int(eta.timestamp())) # we use the received_at param to measure and log our task processing delay. # skip that if we're deliberately rate limiting/delaying the task. params.pop('received_at', None) delay_msg = f'in {eta - now}'
# construct task object body = urllib.parse.urlencode(sorted(params.items()), doseq=True).encode() task = { 'app_engine_http_request': { 'http_method': 'POST', 'relative_uri': path, 'body': body, 'headers': { 'Content-Type': 'application/x-www-form-urlencoded', # propagate trace id. these two headers use different formats, so # pass each through under its own name instead of crossing them. # https://cloud.google.com/trace/docs/trace-context#http-requests # https://stackoverflow.com/a/71343735/186123 'traceparent': traceparent, 'X-Cloud-Trace-Context': cloud_trace, }, }, } if schedule_time: task['schedule_time'] = schedule_time
parent = tasks_client.queue_path(app_id, TASKS_LOCATION, queue) try: task = tasks_client.create_task(parent=parent, task=task) except InvalidArgument as e: if e.message == 'Task size too large': logger.exception('Payload too big') error('Payload too big', 299) raise
msg = f'Added {queue} {task.name.split("/")[-1]} {delay_msg}' if not traceparent and not cloud_trace: logger.info(msg)
return msg, 202
def report_exception(**kwargs): return report_error(msg=None, exception=True, **kwargs)
def report_error(msg, *, exception=False, **kwargs): """Reports an error to StackDriver Error Reporting.
https://cloud.google.com/python/docs/reference/clouderrorreporting/latest/google.cloud.error_reporting.client.Client
If ``DEBUG`` and ``exception`` are ``True``, re-raises the exception instead.
Duplicated in ``bridgy.util``. """ if DEBUG or LOCAL_SERVER: if DEBUG and exception: raise else: # must be at warning level. logging with exception at error level or # above will report to prod error reporting logger.warning(msg, exc_info=exception) return
http_context = build_flask_context(request) if has_request_context() else None
try: if exception: logger.warning(msg, exc_info=True) error_reporting_client.report_exception(http_context=http_context, **kwargs) else: logger.error(msg) error_reporting_client.report(msg, http_context=http_context, **kwargs) except BaseException: kwargs['exception'] = exception logger.warning(f'Failed to report error! {kwargs}', exc_info=exception)
NDB_CONTEXT_KWARGS = { 'cache_policy': memcache.cache_policy, 'global_cache': memcache.global_cache, 'global_cache_policy': memcache.global_cache_policy, 'global_cache_timeout_policy': memcache.global_cache_timeout_policy,}
def log_request(): """Logs GET query params and POST form.
Limits each value to 1000 chars.""" logger.info(f'Params:\n' + '\n'.join( f'{k} = {v[:1000]}' for k, v in request.values.items()))
def secret_key_auth(fn): """Flask decorator that returns HTTP 401 if the request isn't authorized.
Right now this only handles internal authorization: the ``Authorization`` header has to be set to the Flask secret key in the ``flask_secret_key`` file.
Ignored if ``LOCAL_SERVER`` is True.
Must be used *below* :meth:`flask.Flask.route`, eg:
@app.route('/path') @secret_key_auth def handler(): ... """ @functools.wraps(fn) def decorated(*args, **kwargs): if request.headers.get('Authorization') != config.SECRET_KEY: return '', 401
return fn(*args, **kwargs)
return decorated
def make_jwt(*, user, scope, expiration=timedelta(weeks=1), **claims): """Makes a per-user JWT signed by our EncryptedProperty symmetric key.
Args: user (User) scope (str) expiration (timedelta) **claims (str: str): optional additional claims
Returns: str: """ claims.update({ 'sub': user.key.id(), 'scope': scope, 'exp': util.now() + expiration, }) return jwt.encode(claims, key=models.ENCRYPTED_PROPERTY_KEYS_BYTES[0], algorithm='HS256')
def verify_jwt(token, *, user_id, scope, **claims): """Verifies a per-user JWT and checks that it matches a user, scope, etc.
Raises the appropriate werkzeug HTTPException if the JWT doesn't verify or match, otherwise returns None.
Args: token (str) user_id (str) scope (str) **claims (str: str): optional additional claims to check
Raises: werkzeug.exceptions.Unauthorized: if the token is invalid werkzeug.exceptions.Forbidden: if the token is valid but for the wrong user or scope """ decoded = jwt.decode(token, key=models.ENCRYPTED_PROPERTY_KEYS_BYTES[0], algorithms=['HS256'])
for key, expected in list(claims.items()) + [('sub', user_id), ('scope', scope)]: if (got := decoded.get(key)) != expected: raise ValueError(f'expected {key} {expected}, got {got}')
class FlashErrors(View): """Wraps a Flask :class:`flask.view.View` and flashes errors.
Mostly used with OAuth endpoints. """ def dispatch_request(self): try: return super().dispatch_request() except (ValueError, requests.RequestException) as e: logger.warning(f'{self.__class__.__name__} error', exc_info=True) _, body = interpret_http_exception(e) flask_util.flash(body or str(e)) return redirect('/login')
def render_template(template, **kwargs): return flask.render_template( template, isinstance=isinstance, READ_ONLY=READ_ONLY, request=request, set=set, string=string, util=util, **kwargs)
disable_if_read_only = flask_util.disable_if_read_only(render_fn=render_template)