#!/usr/bin/env python3 # Licensed to the Apache Software Foundation (ASF) under one # or more contributor license agreements. See the NOTICE file # distributed with this work for additional information # regarding copyright ownership. The ASF licenses this file # to you under the Apache License, Version 2.0 (the # "License"); you may not use this file except in compliance # with the License. You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, # software distributed under the License is distributed on an # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY # KIND, either express or implied. See the License for the # specific language governing permissions and limitations # under the License. """ASF Infrastructure Reporting Dashboard - Quart Middleware""" if not __debug__: raise RuntimeError("This code requires assert statements to be enabled") import sys import traceback import typing import uuid import quart from . import config import werkzeug.routing import asyncio import functools import aiohttp import time import json async def consume_body(): """Consumes the request body, punting it to dev-null. This is required for httpd to not throw 502 at error""" # See: https://bz.apache.org/bugzilla/show_bug.cgi?id=55433 async for _data in quart.request.body: pass def glued(func: typing.Callable) -> typing.Callable: """Middleware that collects all form data (except file uploads!) and joins as one dict""" async def call(**args): form_data = dict() form_data.update(quart.request.args.to_dict()) xform = await quart.request.form # Pre-parse check for form data size if quart.request.content_type and any( x in quart.request.content_type for x in ( "multipart/form-data", "application/x-www-form-urlencoded", "application/x-url-encoded", ) ): if quart.request.content_length > config.server.max_form_size: await consume_body() return quart.Response( status=413, response=f"Request content length ({quart.request.content_length} bytes) is larger than what is permitted for form data ({config.server.max_form_size} bytes)!", ) if xform: form_data.update(xform.to_dict()) if quart.request.is_json: xjson = await quart.request.json form_data.update(xjson) try: resp = await func(form_data, **args) assert resp, "No response was provided by the underlying endpoint!" except Exception: # Catch and spit out errors exc_type, exc_value, exc_traceback = sys.exc_info() err = "\n".join(traceback.format_exception(exc_type, exc_value, exc_traceback)) headers = { "Server": "ASF Infra Reporting Dashboard", "Content-Type": "text/plain", } # By default, we print the traceback to the user, for easy debugging. if config.server.error_reporting == "json": error_text = "API error occurred: \n" + err return quart.Response(headers=headers, status=500, response=error_text) # If client traceback is disabled, we print it to stderr instead, but leave an # error ID for the client to report back to the admin. Every line of the traceback # will have this error ID at the beginning of the line, for easy grepping. else: # We only need a short ID here, let's pick 18 chars. eid = str(uuid.uuid4())[:18] sys.stderr.write("API Endpoint %s got into trouble (%s): \n" % (quart.request.path, eid)) for line in err.split("\n"): sys.stderr.write("%s: %s\n" % (eid, line)) return quart.Response( headers=headers, status=500, response="API error occurred. The application journal will have information. Error ID: %s" % eid, ) # If an error is thrown before the request body has been consumed, eat it quietly. if not quart.request.body._complete.is_set(): await consume_body() return resp # Quart will, if no rule name is specified, default to calling the rule "call" here, # which leads to carps about duplicate rule definitions. So, given the fact that call() # is dynamically made from within this function, we simply adjust its internal name to # refer to the calling module and function, thus providing Quart with a much better # name for the rule, which will also aid in debugging. call.__name__ = func.__module__ + "." + func.__name__ return call def auth_failed(): """Returns the appropriate authorization failure response, depending on auth mechanism supplied.""" if "x-artifacts-webui" not in quart.request.headers: # Not done via Web UI, standard 401 response headers = {"WWW-Authenticate": 'Basic realm="infra-reports-ec2-va.apache.org"'} return quart.Response(status=401, headers=headers, response="Please authenticate yourself first!\n") else: # Web UI response, do not send Realm header (we do not want a pop-up in our browser!) return quart.Response(status=401, response="Please authenticate yourself first!\n") class FilenameConverter(werkzeug.routing.BaseConverter): """Simple converter that splits a filename into a basename and an extension""" regex = r"^[^/.]*(\.[A-Za-z0-9]+)?$" part_isolating = False def to_python(self, filename): extension = "" # If foo.bar, split into base and ext. Otherwise, keep filename as full string (even for .htaccess etc) if "." in filename[1:]: filename, extension = filename.split(".", maxsplit=1) return filename, extension async def reset_rate_limits(): """Reset daily rate limits for lookups""" while True: await asyncio.sleep(86400) config.rate_limits.clear() def rate_limited(func): """Decorator for calls that are rate-limited for anonymous users. Once the number of requests per day has been exceeded, this decorator will return a 429 HTTP response to the client instead. """ @functools.wraps(func) async def session_wrapper(*args): ip = quart.request.headers.get("X-Forwarded-For", quart.request.remote_addr).split(",")[-1].strip() usage = config.rate_limits.get(ip, 0) + 1 if config.server.rate_limit_per_ip and usage > config.server.rate_limit_per_ip: return quart.Response( status=429, response="Your request has been rate-limited. Please check back tomorrow!" ) config.rate_limits[ip] = usage print(ip, usage) return await func(*args) return session_wrapper class CachedJson: """A simple JSON URL fetcher with a built-in cache. Once a cached JSON element expires, any subsequent reads will cause the class to re-fetch the object. If the object cannot be fetched, a stale version will be returned instead. Usage example: foo = CachedJson("https://www.apache.org/foo.json") jsondata = await foo.json # do stuff here.... jsondata = await foo.json # <- will either use cache or fetch again if out of date """ def __init__(self, url: str, expiry: int = 30): self.url = url self.expiry = expiry self.last = 0 self.timeout = aiohttp.ClientTimeout(total=30) self._cache = None @property async def json(self): now = time.time() if now > (self.last + self.expiry): # Cache expired? try: async with aiohttp.ClientSession(timeout=self.timeout) as hc: async with hc.get(self.url) as req: if req.status == 200: source_json = await req.json() if source_json: self._cache = source_json self.last = now except (aiohttp.ClientError, json.JSONDecodeError) as e: print(f"Could not fetch URL {self.url} for caching, will use stale checkout: {e}.") return self._cache