# 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. from __future__ import annotations import asyncio import functools import logging import time from typing import TYPE_CHECKING, Any, Final, NoReturn, ParamSpec, Protocol, TypeVar import aiofiles import aiofiles.os import asfquart import asfquart.auth as auth import asfquart.base as base import asfquart.session as session import quart import atr.db as db import atr.db.models as models import atr.user as user import atr.util as util from atr import config if TYPE_CHECKING: from collections.abc import Awaitable, Callable, Coroutine, Sequence import werkzeug.datastructures as datastructures import werkzeug.wrappers.response as response if asfquart.APP is ...: raise RuntimeError("APP is not set") P = ParamSpec("P") R = TypeVar("R", covariant=True) T = TypeVar("T") # TODO: Should get this from config, checking debug there _MEASURE_PERFORMANCE: Final[bool] = True _LOGGER: Final = logging.getLogger(__name__) # | 1 | RSA (Encrypt or Sign) [HAC] | # | 2 | RSA Encrypt-Only [HAC] | # | 3 | RSA Sign-Only [HAC] | # | 16 | Elgamal (Encrypt-Only) [ELGAMAL] [HAC] | # | 17 | DSA (Digital Signature Algorithm) [FIPS186] [HAC] | # | 18 | ECDH public key algorithm | # | 19 | ECDSA public key algorithm [FIPS186] | # | 20 | Reserved (formerly Elgamal Encrypt or Sign) | # | 21 | Reserved for Diffie-Hellman | # | | (X9.42, as defined for IETF-S/MIME) | # | 22 | EdDSA [I-D.irtf-cfrg-eddsa] | # - https://lists.gnupg.org/pipermail/gnupg-devel/2017-April/032762.html # TODO: (Obviously we should move this, but where to?) algorithms: Final[dict[int, str]] = { 1: "RSA", 2: "RSA", 3: "RSA", 16: "Elgamal", 17: "DSA", 18: "ECDH", 19: "ECDSA", 21: "Diffie-Hellman", 22: "EdDSA", } class AsyncFileHandler(logging.Handler): """A logging handler that writes logs asynchronously using aiofiles.""" def __init__(self, filename, mode="w", encoding=None): super().__init__() self.filename = filename if mode != "w": raise RuntimeError("Only write mode is supported") self.encoding = encoding self.queue = asyncio.Queue() self.our_worker_task = None def our_worker_task_ensure(self): """Lazily create the worker task if it doesn't exist and there's an event loop.""" if self.our_worker_task is None: try: loop = asyncio.get_running_loop() self.our_worker_task = loop.create_task(self.our_worker()) except RuntimeError: # No event loop running yet, try again on next emit ... async def our_worker(self): """Background task that writes queued log messages to file.""" # Use a binary mode literal with aiofiles.open # https://github.com/Tinche/aiofiles/blob/main/src/aiofiles/threadpool/__init__.py # We should be able to use any mode, but pyright requires a binary mode async with aiofiles.open(self.filename, "wb+") as f: while True: record = await self.queue.get() if record is None: break try: # Format the log record first formatted_message = self.format(record) + "\n" message_bytes = formatted_message.encode(self.encoding or "utf-8") await f.write(message_bytes) await f.flush() except Exception: self.handleError(record) finally: self.queue.task_done() def emit(self, record): """Queue the record for writing by the worker task.""" try: # Ensure worker task is running self.our_worker_task_ensure() # Queue the record, but handle the case where no event loop is running yet try: self.queue.put_nowait(record) except RuntimeError: ... except Exception: self.handleError(record) def close(self): """Shut down the worker task cleanly.""" if self.our_worker_task is not None and not self.our_worker_task.done(): try: self.queue.put_nowait(None) except RuntimeError: # No running event loop, no need to clean up ... super().close() # This is the type of functions to which we apply @committer_get # In other words, functions which accept CommitterSession as their first arg class CommitterRouteHandler(Protocol[R]): """Protocol for @committer_get decorated functions.""" __name__: str __doc__: str | None def __call__(self, session: CommitterSession, *args: Any, **kwargs: Any) -> Awaitable[R]: ... class CommitterSession: """Session with extra information about committers.""" def __init__(self, web_session: session.ClientSession) -> None: self._projects: list[models.Project] | None = None self._session = web_session def __getattr__(self, name: str) -> Any: # TODO: Not type safe, should subclass properly if possible # For example, we can access session.no_such_attr and the type checkers won't notice return getattr(self._session, name) async def check_access(self, project_name: str) -> None: if not any((p.name == project_name) for p in (await self.user_projects)): raise base.ASFQuartException("You do not have access to this project", errorcode=403) @property def app_host(self) -> str: return config.get().APP_HOST @property def host(self) -> str: request_host = quart.request.host if ":" in request_host: domain, port = request_host.split(":") # Could be an IPv6 address, so need to check whether port is a valid integer if port.isdigit(): return domain return request_host def only_user_releases(self, releases: Sequence[models.Release]) -> list[models.Release]: return util.user_releases(self.uid, releases) async def redirect( self, route: CommitterRouteHandler[R], success: str | None = None, error: str | None = None, **kwargs: Any ) -> response.Response: """Redirect to a route with a success or error message.""" if success is not None: await quart.flash(success, "success") elif error is not None: await quart.flash(error, "error") return quart.redirect(util.as_url(route, **kwargs)) async def release( self, project_name: str, version_name: str, phase: models.ReleasePhase | db.NotSet | None = db.NOT_SET, data: db.Session | None = None, with_committee: bool = False, with_packages: bool = False, with_project: bool = True, with_tasks: bool = False, ) -> models.Release: # We reuse db.NOT_SET as an entirely different sentinel # TODO: We probably shouldn't do that, or should make it clearer if phase is None: phase_value = db.NOT_SET elif phase is db.NOT_SET: phase_value = models.ReleasePhase.RELEASE_CANDIDATE_DRAFT else: phase_value = phase release_name = models.release_name(project_name, version_name) if data is None: async with db.session() as data: release = await data.release( name=release_name, phase=phase_value, _committee=with_committee, _project=with_project, _tasks=with_tasks, ).demand(base.ASFQuartException("Release does not exist", errorcode=404)) else: release = await data.release( name=release_name, phase=phase_value, _committee=with_committee, _project=with_project, _tasks=with_tasks, ).demand(base.ASFQuartException("Release does not exist", errorcode=404)) return release @property async def user_candidate_drafts(self) -> list[models.Release]: return await user.candidate_drafts(self.uid, user_projects=self._projects) # @property # async def user_committees(self) -> list[models.Committee]: # return ... @property async def user_projects(self) -> list[models.Project]: if self._projects is None: self._projects = await user.projects(self.uid) return self._projects @property async def user_releases(self) -> list[models.Release]: return await user.releases(self.uid) class FlashError(RuntimeError): ... class MicrosecondsFormatter(logging.Formatter): # Answers on a postcard if you know why Python decided to use a comma by default default_msec_format = "%s.%03d" # Setup a dedicated logger for route performance metrics # NOTE: This code block must come after AsyncFileHandler and MicrosecondsFormatter route_logger: Final = logging.getLogger("route.performance") # Use custom formatter that properly includes microseconds # TODO: Is this actually UTC? route_logger_handler: Final[AsyncFileHandler] = AsyncFileHandler("route-performance.log") route_logger_handler.setFormatter(MicrosecondsFormatter("%(asctime)s - %(message)s")) route_logger.addHandler(route_logger_handler) route_logger.setLevel(logging.INFO) # If we don't set propagate to False then it logs to the term as well route_logger.propagate = False # This is the type of functions to which we apply @app_route # In other words, functions which accept no session class RouteHandler(Protocol[R]): """Protocol for @app_route decorated functions.""" __name__: str __doc__: str | None def __call__(self, *args: Any, **kwargs: Any) -> Awaitable[R]: ... def app_route( path: str, methods: list[str] | None = None, endpoint: str | None = None, measure_performance: bool = True ) -> Callable: """Register a route with the Flask app with built-in performance logging.""" def decorator(f: Callable[P, Coroutine[Any, Any, T]]) -> Callable[P, Awaitable[T]]: # First apply our performance measuring decorator if _MEASURE_PERFORMANCE and measure_performance: measured_func = app_route_performance_measure(path, methods)(f) else: measured_func = f # Then apply the original route decorator return asfquart.APP.route(path, methods=methods, endpoint=endpoint)(measured_func) return decorator def app_route_performance_measure(route_path: str, http_methods: list[str] | None = None) -> Callable: """Decorator that measures and logs route performance with path and method information.""" # def format_time(seconds: float) -> str: # """Format time in appropriate units (µs or ms).""" # microseconds = seconds * 1_000_000 # if microseconds < 1000: # return f"{microseconds:.2f} µs" # else: # milliseconds = microseconds / 1000 # return f"{milliseconds:.2f} ms" def decorator(f: Callable[P, Coroutine[Any, Any, T]]) -> Callable[P, Awaitable[T]]: @functools.wraps(f) async def wrapper(*args: P.args, **kwargs: P.kwargs) -> T: # This wrapper is based on an outstanding idea by Mostafa Farzán # Farzán realised that we can step the event loop manually # That way, we can also divide it into synchronous and asynchronous parts # The synchronous part is done using coro.send(None) # The asynchronous part is done using asyncio.sleep(0) # We use two methods for measuring the async part, and take the largest # This performance measurement adds a bit of overhead, about 10-20ms # Therefore it should be avoided in production, or made more efficient # We could perhaps use for a small portion of requests blocking_time = 0.0 async_time = 0.0 loop_time = 0.0 total_start = time.perf_counter() coro = f(*args, **kwargs) try: while True: # Measure the synchronous part sync_start = time.perf_counter() future = coro.send(None) sync_end = time.perf_counter() blocking_time += sync_end - sync_start # Measure the asynchronous part in two different ways loop = asyncio.get_running_loop() wait_start = time.perf_counter() loop_start = loop.time() if future is not None: while not future.done(): await asyncio.sleep(0) wait_end = time.perf_counter() loop_end = loop.time() async_time += wait_end - wait_start loop_time += loop_end - loop_start # Raise exception if any # future.result() except StopIteration as e: total_end = time.perf_counter() total_time = total_end - total_start methods_str = ",".join(http_methods) if http_methods else "GET" nonblocking_time = max(async_time, loop_time) # If async time is more than 10% different from loop time, log it delta_symbol = "=" nonblocking_delta = abs(async_time - loop_time) # Must check that nonblocking_time is not 0 to avoid division by zero if nonblocking_time and ((nonblocking_delta / nonblocking_time) > 0.1): delta_symbol = "!" route_logger.info( "%s %s %s %s %s %s %s", methods_str, route_path, f.__name__, delta_symbol, int(blocking_time * 1000), int(nonblocking_time * 1000), int(total_time * 1000), ) return e.value return wrapper return decorator # This decorator is an adaptor between @committer_get and @app_route functions def committer( path: str, methods: list[str] | None = None, measure_performance: bool = True ) -> Callable[[CommitterRouteHandler[R]], RouteHandler[R]]: """Decorator for committer GET routes that provides an enhanced session object.""" def decorator(func: CommitterRouteHandler[R]) -> RouteHandler[R]: async def wrapper(*args: Any, **kwargs: Any) -> R: web_session = await session.read() if web_session is None: _authentication_failed() enhanced_session = CommitterSession(web_session) return await func(enhanced_session, *args, **kwargs) # Generate a unique endpoint name endpoint = func.__module__ + "_" + func.__name__ # Set the name before applying decorators wrapper.__name__ = func.__name__ wrapper.__doc__ = func.__doc__ wrapper.__annotations__["endpoint"] = endpoint # Apply decorators in reverse order decorated = auth.require(auth.Requirements.committer)(wrapper) decorated = app_route( path, methods=methods or ["GET"], endpoint=endpoint, measure_performance=measure_performance )(decorated) return decorated return decorator async def get_form(request: quart.Request) -> datastructures.MultiDict: # The request.form() method in Quart calls a synchronous tempfile method # It calls quart.wrappers.request.form _load_form_data # Which calls quart.formparser parse and parse_func and parser.parse # Which calls _write which calls tempfile, which is synchronous # It's getting a tempfile back from some prior call # We can't just make blockbuster ignore the call because then it ignores it everywhere app = asfquart.APP if app is ...: raise RuntimeError("APP is not set") # Or quart.current_app? blockbuster = app.extensions.get("blockbuster") # Turn blockbuster off if blockbuster is not None: blockbuster.deactivate() form = await request.form # Turn blockbuster on if blockbuster is not None: blockbuster.activate() return form def public( path: str, methods: list[str] | None = None, measure_performance: bool = True ) -> Callable[[RouteHandler[R]], RouteHandler[R]]: """Decorator for public GET routes that provides an enhanced session object.""" def decorator(func: RouteHandler[R]) -> RouteHandler[R]: async def wrapper(*args: Any, **kwargs: Any) -> R: return await func(*args, **kwargs) # Generate a unique endpoint name endpoint = func.__module__ + "_" + func.__name__ # Set the name before applying decorators wrapper.__name__ = func.__name__ wrapper.__doc__ = func.__doc__ wrapper.__annotations__["endpoint"] = endpoint # Apply decorators in reverse order decorated = app_route( path, methods=methods or ["GET"], endpoint=endpoint, measure_performance=measure_performance )(wrapper) return decorated return decorator def _authentication_failed() -> NoReturn: """Handle authentication failure with an exception.""" # NOTE: This is a separate function to fix a problem with analysis flow in mypy raise base.ASFQuartException("Not authenticated", errorcode=401)