| #!/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 |
