diff --git a/py2http/auth.py b/py2http/auth.py new file mode 100644 index 0000000..7756fc9 --- /dev/null +++ b/py2http/auth.py @@ -0,0 +1,174 @@ +"""Framework-independent credential checks, shared by the aiohttp middleware +(``py2http.middleware``) and the Bottle plugins (``py2http.bottle_plugins``), so that +both refuse the same tokens and keys for the same reasons. + +>>> bearer_token('Bearer abc'), bearer_token('bearer abc'), bearer_token('Basic abc') +('abc', 'abc', '') +>>> key_matches('k-123', 'k-123'), key_matches('k-12', 'k-123'), key_matches(None, 'k') +(True, False, False) +""" + +import hmac +import logging +from collections.abc import Iterable, Mapping +from typing import Optional + +logger = logging.getLogger("py2http") + +DFLT_JWT_ALGORITHMS = ("HS256",) +# ``jwt.decode`` arguments the verification setup owns, so ``decode_kwargs`` can't +_RESERVED_DECODE_KWARGS = frozenset({"jwt", "key", "algorithms", "options"}) +_BEARER_PREFIX = "bearer " + + +class CredentialsRejected(Exception): + """A request's credentials were refused. The message is safe to show the client: + it never contains the token, nor anything about the server's key.""" + + +def _algorithm_family(algorithm: str) -> str: + """The key family an algorithm verifies with: a shared secret for ``HS*``, a + public key otherwise. + + >>> _algorithm_family('HS512'), _algorithm_family('RS256'), _algorithm_family('ES256') + ('hmac', 'public-key', 'public-key') + """ + return "hmac" if algorithm.upper().startswith("HS") else "public-key" + + +def _names(name: str, value: Iterable[str]) -> list: + if isinstance(value, (str, bytes)): + raise TypeError(f"{name} must be a list of names, e.g. [{value!r}]") + return list(value) + + +def jwt_algorithms( + secret, + *, + verify: bool = True, + algorithms: Iterable[str] = DFLT_JWT_ALGORITHMS, + allow_mixed_algorithm_families: bool = False, +) -> list: + """Validate a JWT verification setup, returning the algorithms as a list. + + Raises at construction time, rather than accepting forgeries at request time, when: + + - ``algorithms`` is empty, a bare string, or names an unknown or unsigned (``none``) + algorithm (PyJWT names are case-sensitive); + - ``verify`` is True and ``secret`` is empty (an empty HMAC secret lets anyone sign + tokens), or an HMAC algorithm is asked for with a non-``str``/``bytes`` secret; + - ``verify`` is True and HMAC and public-key algorithms are mixed with one + ``secret`` (the classic algorithm-confusion setup), unless + ``allow_mixed_algorithm_families`` is True. + + >>> jwt_algorithms('s3cret') + ['HS256'] + >>> jwt_algorithms('', algorithms=['HS256']) + Traceback (most recent call last): + ... + ValueError: secret must be a non-empty key when verify=True + """ + import jwt + + algorithms = _names("algorithms", algorithms) + if not algorithms: + raise ValueError("algorithms must name at least one JWT algorithm") + known = set(jwt.algorithms.get_default_algorithms()) - {"none"} + if unknown := [a for a in algorithms if a not in known]: + raise ValueError( + f"Unknown or unsigned JWT algorithm(s) {unknown}; names are " + f"case-sensitive, choose from {sorted(known)}" + ) + if verify and not secret: + raise ValueError("secret must be a non-empty key when verify=True") + families = {_algorithm_family(a) for a in algorithms} + if verify and "hmac" in families and not isinstance(secret, (str, bytes)): + raise TypeError(f"an HMAC secret must be str or bytes, got {type(secret)}") + if verify and len(families) > 1 and not allow_mixed_algorithm_families: + raise ValueError( + f"algorithms={algorithms} mixes HMAC and public-key algorithms with one " + "secret, which allows algorithm-confusion forgeries; use one family, or " + "pass allow_mixed_algorithm_families=True if you really mean it" + ) + return algorithms + + +def jwt_decode_kwargs(decode_kwargs: Optional[Mapping] = None) -> dict: + """Validate extra ``jwt.decode`` arguments (``audience``, ``issuer``, + ``leeway``...), refusing those the verification setup owns. + + >>> jwt_decode_kwargs({'audience': 'my-api'}) + {'audience': 'my-api'} + >>> jwt_decode_kwargs({'options': {'verify_signature': False}}) + Traceback (most recent call last): + ... + ValueError: decode_kwargs may not set ['options']; use verify= and algorithms= + """ + decode_kwargs = dict(decode_kwargs or {}) + if reserved := sorted(_RESERVED_DECODE_KWARGS & set(decode_kwargs)): + raise ValueError( + f"decode_kwargs may not set {reserved}; use verify= and algorithms=" + ) + return decode_kwargs + + +def bearer_token(authorization_header) -> str: + """The token of an ``Authorization: Bearer `` header, or ``''``.""" + header = authorization_header or "" + if header[: len(_BEARER_PREFIX)].lower() != _BEARER_PREFIX: + return "" + return header[len(_BEARER_PREFIX) :].strip() + + +def decode_jwt( + token: str, + secret, + *, + algorithms: list, + verify: bool = True, + decode_kwargs: Optional[Mapping] = None, +) -> dict: + """The claims of ``token``, or ``CredentialsRejected`` with a client-safe reason. + + With ``verify=False`` neither the signature nor the claims (``exp``...) are checked. + ``decode_kwargs`` (``audience``, ``issuer``, ``leeway``...) go to ``jwt.decode``. + """ + import jwt + + try: + return jwt.decode( + token, + secret, + options={"verify_signature": verify}, + algorithms=algorithms, + **(decode_kwargs or {}), + ) + except jwt.InvalidTokenError as error: # its text describes the token + raise CredentialsRejected(f"Invalid authentication token: {error}") from error + except (jwt.PyJWTError, UnicodeError) as error: + # a key unfit for the token's alg (InvalidKeyError...) describes the server's + # key, so it is logged rather than sent; a non-UTF-8 header is just refused + if not isinstance(error, UnicodeError): + logger.warning("JWT rejected for a server-side reason: %r", error) + raise CredentialsRejected( + "Invalid authentication token: not acceptable" + ) from error + + +def required_key(key, *, name: str = "api_key"): + """``key``, refused when empty: an empty key would let a request with no + ``Authorization`` header through.""" + if not key or not isinstance(key, (str, bytes)): + raise ValueError(f"{name} must be a non-empty str or bytes") + return key + + +def key_matches(given, expected) -> bool: + """Whether the ``given`` credential equals ``expected``, in constant time.""" + if not given or not expected: + return False + + def _bytes(x): # aiohttp keeps non-UTF-8 header bytes as surrogate escapes + return x.encode(errors="surrogateescape") if isinstance(x, str) else x + + return hmac.compare_digest(_bytes(given), _bytes(expected)) diff --git a/py2http/bottle_plugins.py b/py2http/bottle_plugins.py index df92146..827bc80 100644 --- a/py2http/bottle_plugins.py +++ b/py2http/bottle_plugins.py @@ -1,16 +1,42 @@ """Plugins for adding middleware functionality to Bottle apps.""" -from bottle import request, response, abort -from functools import wraps import json -import jwt +from functools import wraps from typing import Iterable from warnings import warn + +from bottle import request, response + +from py2http.auth import ( + DFLT_JWT_ALGORITHMS, + CredentialsRejected, + bearer_token, + decode_jwt, + jwt_algorithms, + jwt_decode_kwargs, + key_matches, + required_key, +) from py2http.constants import JSON_CONTENT_TYPE OPTIONS = "OPTIONS" +def _authorization_header() -> str: + """The request's ``Authorization`` header; ``''`` if it is not valid UTF-8 (Bottle + raises when decoding it), which every check then refuses.""" + try: + return request.headers.get("Authorization", "") + except UnicodeError: + return "" + + +def _unauthorized(message): + response.status = 401 + response.content_type = JSON_CONTENT_TYPE + return json.dumps({"error": message}) + + class JWTPlugin: """A plugin for validating JWTs and extracting their payloads. @@ -25,70 +51,82 @@ def __init__( mapper: dict = None, ignore_methods: Iterable[str] = None, algorithms: Iterable[str] = None, + *, + allow_mixed_algorithm_families: bool = False, + decode_kwargs: dict = None, ): """Creates a new JWTPlugin instance. - :param secret: (Optional) The JWT public key (RS256) or synchronous secret (HS256) used to validate tokens. + :param secret: The JWT public key (RS256) or shared secret (HS256) used to + validate tokens. Required (non-empty) when ``verify`` is True. :param verify: (Optional) If True, will verify JWT signatures against the provided secret, and reject unverified requests. :param mapper: (Optional) A dict that specifies how to map JWT claim value names in the output. :param ignore_methods: (Optional) A list of method names for the plugin to ignore. + :param algorithms: (Optional) The algorithms a token may be signed with + (default ``("HS256",)``); validated as ``py2http.auth.jwt_algorithms`` + describes, as is ``allow_mixed_algorithm_families``. + :param decode_kwargs: (Optional) Extra ``jwt.decode`` arguments, such as + ``audience``, ``issuer`` or ``leeway``. """ self._secret = secret self._verify = verify self._mapper = mapper if mapper else {} self._ignore_methods = ignore_methods if ignore_methods else [] - self._algorithms = algorithms if algorithms else ["HS256"] + self._algorithms = jwt_algorithms( + secret, + verify=verify, + algorithms=algorithms or DFLT_JWT_ALGORITHMS, + allow_mixed_algorithm_families=allow_mixed_algorithm_families, + ) + self._decode_kwargs = jwt_decode_kwargs(decode_kwargs) + + def _claims(self): + """The (renamed) claims of the request's token; raises ``CredentialsRejected``.""" + decoded = decode_jwt( + bearer_token(_authorization_header()), + self._secret, + algorithms=self._algorithms, + verify=self._verify, + decode_kwargs=self._decode_kwargs, + ) + for k, v in self._mapper.items(): + if k in decoded: + decoded[v] = decoded.pop(k) + return decoded def __call__(self, handler): - if self._ignore_methods and handler.method_name in self._ignore_methods: + if getattr(handler, "method_name", None) in self._ignore_methods: return handler @wraps(handler) def wrapped_handler(*args, **kwargs): if request.method == OPTIONS: - return handler(*args, *kwargs) - auth_header = request.headers.get("Authorization", "") - token = auth_header[7:] - try: - decoded = jwt.decode( - token, - self._secret, - options={"verify_signature": self._verify}, - algorithms=self._algorithms, - ) - for k, v in self._mapper.items(): - if k in decoded: - decoded[v] = decoded.pop(k) - request.token = decoded return handler(*args, **kwargs) - except jwt.DecodeError as error: + try: + request.token = self._claims() + except CredentialsRejected as rejection: if self._verify: - response.status = 401 - response.content_type = JSON_CONTENT_TYPE - return json.dumps( - { - "error": f'Invalid authentication token "{token}", {str(error)}' - } - ) - warn(f"Invalid JWT: {token}") - return handler(*args, **kwargs) + return _unauthorized(str(rejection)) + warn(f"Invalid JWT: {rejection.__cause__}") + return handler(*args, **kwargs) return wrapped_handler class ApiKeyAuthPlugin: + """A plugin letting through only requests whose ``Authorization`` header is + exactly ``api_key`` (compared in constant time; an empty key is refused).""" + def __init__(self, api_key: str): - self._api_key = api_key + self._api_key = required_key(api_key) def __call__(self, handler): + @wraps(handler) def wrapped_handler(*args, **kwargs): - auth_header = request.headers.get("Authorization", "") - if auth_header == self._api_key: + if key_matches(_authorization_header(), self._api_key): return handler(*args, **kwargs) - response.status = 401 - response.content_type = JSON_CONTENT_TYPE - return json.dumps({"error": "invalid API key"}) + return _unauthorized("invalid API key") return wrapped_handler @@ -100,6 +138,7 @@ def __init__(self, origins: str = "*"): self._origins = origins def __call__(self, handler): + @wraps(handler) def wrapped_handler(*args, **kwargs): response.headers["Access-Control-Allow-Origin"] = self._origins response.headers["Access-Control-Allow-Methods"] = ( diff --git a/py2http/middleware.py b/py2http/middleware.py index a2ee5d8..efe8ba3 100644 --- a/py2http/middleware.py +++ b/py2http/middleware.py @@ -1,41 +1,35 @@ -""" -py2http is a Python module that allows you to dispatch Python functions as HTTP -services. -By running a basic HTTP service from a list of functions, you can access OpenAPI -specifications, Swagger documentation, and utilize the HTTP service's routes. -Additionally, Py2http provides features for method transformation, input mapping, -output mapping, error handling, and client generation. - -The code snippet provided below is a part of Py2http module and includes middleware -functions for handling JWT authentication and superadmin authorization. -These middleware functions are designed to be used within the Py2http framework to -ensure secure and authenticated access to HTTP routes. - -Include the following docstring to describe this module: -"Middleware functions for handling JWT authentication and superadmin authorization -within the Py2http framework. These functions ensure secure access to HTTP routes -by verifying credentials and permissions. Use these middleware functions to enforce -authentication and control access to web resources. -""" +"""aiohttp middleware that guards every route of an app: ``mk_jwt_middleware`` +(a verified ``Authorization: Bearer`` JWT) and ``mk_superadmin_middleware`` (one +shared key). The checks themselves live in ``py2http.auth``, shared with the Bottle +plugins of ``py2http.bottle_plugins``.""" import json from collections.abc import Iterable from warnings import warn +from py2http.auth import ( + DFLT_JWT_ALGORITHMS, + CredentialsRejected, + _algorithm_family, # noqa: F401 (was defined here) + _names, + bearer_token, + decode_jwt, + jwt_algorithms, + jwt_decode_kwargs, + key_matches, + required_key, +) from py2http.constants import JSON_CONTENT_TYPE -DFLT_JWT_ALGORITHMS = ("HS256",) DFLT_PUBLIC_ROUTES = ("ping", "openapi") -def _algorithm_family(algorithm: str) -> str: - """The key family an algorithm verifies with: a shared secret for ``HS*``, a - public key otherwise. +def _unauthorized(message): + from aiohttp import web - >>> _algorithm_family('HS512'), _algorithm_family('RS256'), _algorithm_family('ES256') - ('hmac', 'public-key', 'public-key') - """ - return "hmac" if algorithm.upper().startswith("HS") else "public-key" + return web.HTTPUnauthorized( + text=json.dumps({"error": message}), content_type=JSON_CONTENT_TYPE + ) def mk_jwt_middleware( @@ -45,6 +39,7 @@ def mk_jwt_middleware( algorithms: Iterable[str] = DFLT_JWT_ALGORITHMS, public_routes: Iterable[str] = DFLT_PUBLIC_ROUTES, allow_mixed_algorithm_families: bool = False, + decode_kwargs: dict = None, ): """Make an aiohttp middleware that requires a valid ``Authorization: Bearer`` JWT on every request, and puts the decoded claims on ``request.token``. @@ -71,84 +66,59 @@ def mk_jwt_middleware( public key as the HMAC secret), so it raises ``ValueError`` unless this is True. The middleware used to accept ``["HS256", "RS256"]``; pass that list with this flag to restore it. + decode_kwargs: Extra ``jwt.decode`` arguments, such as ``audience``, + ``issuer`` or ``leeway`` (without them a token carrying an ``aud`` + claim is always refused). """ - import jwt from aiohttp import web - for name, value in [("algorithms", algorithms), ("public_routes", public_routes)]: - if isinstance(value, (str, bytes)): - raise TypeError(f"{name} must be a list of names, e.g. [{value!r}]") - algorithms = list(algorithms) - if not algorithms: - raise ValueError("algorithms must name at least one JWT algorithm") - known = set(jwt.algorithms.get_default_algorithms()) - {"none"} - if unknown := [a for a in algorithms if a not in known]: - raise ValueError( - f"Unknown or unsigned JWT algorithm(s) {unknown}; names are " - f"case-sensitive, choose from {sorted(known)}" - ) - if verify and not secret: - raise ValueError("secret must be a non-empty key when verify=True") - families = {_algorithm_family(a) for a in algorithms} - if verify and "hmac" in families and not isinstance(secret, (str, bytes)): - raise TypeError(f"an HMAC secret must be str or bytes, got {type(secret)}") - if len(families) > 1 and not allow_mixed_algorithm_families: - raise ValueError( - f"algorithms={algorithms} mixes HMAC and public-key algorithms with one " - "secret, which allows algorithm-confusion forgeries; use one family, or " - "pass allow_mixed_algorithm_families=True if you really mean it" - ) - public_routes = frozenset(name for name in public_routes if name is not None) - - def _unauthorized(message): - return web.HTTPUnauthorized( - text=json.dumps({"error": message}), - content_type=JSON_CONTENT_TYPE, - ) + algorithms = jwt_algorithms( + secret, + verify=verify, + algorithms=algorithms, + allow_mixed_algorithm_families=allow_mixed_algorithm_families, + ) + decode_kwargs = jwt_decode_kwargs(decode_kwargs) + public_routes = frozenset( + name for name in _names("public_routes", public_routes) if name is not None + ) @web.middleware async def middleware(req, handler): route = req.match_info.route if getattr(route, "name", None) in public_routes: return await handler(req) - auth_header = req.headers.get("Authorization", "") - token = auth_header[7:] if auth_header[:7].lower() == "bearer " else "" + token = bearer_token(req.headers.get("Authorization")) try: - decoded = jwt.decode( + req.token = decode_jwt( token, secret, - options={"verify_signature": verify}, algorithms=algorithms, + verify=verify, + decode_kwargs=decode_kwargs, ) - except jwt.PyJWTError as error: # incl. a key unfit for the token's alg + except CredentialsRejected as rejection: if verify: - # InvalidTokenError texts describe the token ("Signature has - # expired"); other PyJWTErrors can describe the server's key - reason = ( - str(error) - if isinstance(error, jwt.InvalidTokenError) - else "not acceptable" - ) - raise _unauthorized(f"Invalid authentication token: {reason}") - warn(f"Invalid JWT: {error}") - return await handler(req) - req.token = decoded + raise _unauthorized(str(rejection)) from None + warn(f"Invalid JWT: {rejection.__cause__}") return await handler(req) return middleware def mk_superadmin_middleware(secret): + """Make an aiohttp middleware that only lets through requests whose + ``Authorization`` header is exactly ``secret`` (compared in constant time). + An empty ``secret`` is refused, since it would let header-less requests through. + """ from aiohttp import web + secret = required_key(secret, name="secret") + @web.middleware async def middleware(req, handler): - auth_header = req.headers.get("Authorization", "") - if auth_header == secret: + if key_matches(req.headers.get("Authorization"), secret): return await handler(req) - return web.HTTPUnauthorized( - text=json.dumps({"error": "invalid API key"}), - content_type=JSON_CONTENT_TYPE, - ) + raise _unauthorized("invalid API key") return middleware diff --git a/py2http/obj_dispatch.py b/py2http/obj_dispatch.py new file mode 100644 index 0000000..ff377d0 --- /dev/null +++ b/py2http/obj_dispatch.py @@ -0,0 +1,603 @@ +"""Dispatching HTTP requests to the attributes of objects, classes and modules. + +A handler whose ``endpoint`` is not a function names the attributes clients may call +with ``attr_names`` (see ``public_attr`` for ``'*'``): + +- a **module**: its functions are called by name; +- an **instance** (one object every client shares): ``_instance_endpoint_dispatcher``; +- a **class** (each client creates and uses its own objects): + ``_class_endpoint_dispatcher``. + +Instances and classes need an ``obj_owner``, a ``(request) -> owner`` hook reading +``current_request``, and are bounded in size (``py2http.sizeof``). +""" + +import inspect +import threading +from collections import OrderedDict +from collections.abc import MutableMapping +from contextvars import ContextVar +from types import FunctionType, ModuleType +from typing import Any, Callable, Optional +from uuid import uuid4 + +from i2 import name_of_obj +from i2.errors import AuthorizationError, InputError + +from py2http.sizeof import bounded_deep_sizeof +from py2http.util import call_with_inputs + +# Instance endpoints (``handler['endpoint']`` is an object) are one object every client +# shares; they too need an ``obj_owner`` and are bounded by ``max_obj_size`` (see +# ``_instance_endpoint_dispatcher``). +# +# Class endpoints (``handler['endpoint']`` is a class) keep the objects clients create. +# Who may use an object is the handler's ``obj_owner`` (see ``ObjOwner``); how many +# objects an owner may keep is ``obj_quota``; how many bytes each object, each owner and +# the whole endpoint may hold are ``max_obj_size``, ``max_owner_bytes`` and +# ``max_total_bytes`` (see ``_class_endpoint_dispatcher``). +# +# Maximum number of objects a class endpoint keeps alive in ``shared_objects`` mode; the +# least recently used one is evicted beyond that. Pass ``obj_store=...`` to change it. +DFLT_OBJ_STORE_MAX_SIZE = 1000 +DFLT_OBJ_QUOTA = 100 # objects per owner +DFLT_MAX_OWNERS = 1000 # owners holding objects at once (least recently active evicted) +DFLT_MAX_OBJ_SIZE = 1_000_000 # bytes per object, as measured by ``obj_sizeof`` +DFLT_MAX_OWNER_BYTES = 20_000_000 # bytes per owner (all their objects) +DFLT_MAX_TOTAL_BYTES = 256_000_000 # bytes per endpoint (every object of every owner) + +#: The request being handled (set around every handler call), so per-request hooks such +#: as ``obj_owner`` can read it without the dispatched function taking it. +current_request: ContextVar = ContextVar("py2http_current_request", default=None) + + +def _at_least_one(name: str, value: int) -> int: + if value < 1: + raise ValueError(f"{name} must be at least 1, got {value}") + return value + + +class LRUObjStore(MutableMapping): + """A mapping that keeps at most ``max_size`` items, evicting the least recently + used one when a new key would exceed it (``on_evict(k, v)`` is told). + + Used as the default store of objects created through a class endpoint, so that + clients cannot grow server memory without bound by creating objects. + + >>> s = LRUObjStore(max_size=2) + >>> s['a'], s['b'] = 1, 2 + >>> _ = s['a'] # 'a' is now the most recently used + >>> s['c'] = 3 + >>> sorted(s) + ['a', 'c'] + """ + + def __init__( + self, + *, + max_size: int = DFLT_OBJ_STORE_MAX_SIZE, + on_evict: Optional[Callable[[Any, Any], None]] = None, + ): + self.max_size = _at_least_one("max_size", max_size) + self.on_evict = on_evict + self._data = OrderedDict() + + def __getitem__(self, k): + v = self._data[k] + self._data.move_to_end(k) + return v + + def __setitem__(self, k, v): + self._data[k] = v + self._data.move_to_end(k) + while len(self._data) > self.max_size: + evicted_k, evicted_v = self._data.popitem(last=False) + if self.on_evict is not None: + self.on_evict(evicted_k, evicted_v) + + def __delitem__(self, k): + del self._data[k] + + def peek(self, k, default=None): + """``self.get(k, default)``, without making ``k`` the most recently used.""" + return self._data.get(k, default) + + def __iter__(self): + return iter(self._data) + + def __len__(self): + return len(self._data) + + +class _SharedObjects: + """``obj_owner=shared_objects``: objects belong to no one. Any client holding an + object id may use it, and every client's objects share one LRU store, so a client + creating objects evicts everyone else's. Only for trusted clients.""" + + def __repr__(self): + return "shared_objects" + + +shared_objects = _SharedObjects() + +#: ``obj_owner`` hook: ``(request) -> owner``, a hashable that identifies an +#: *authenticated* client (e.g. a user id from a verified token). ``None`` and ``""`` +#: mean "not identified" and are refused. A client-chosen value (a session cookie the +#: client mints, an IP) lets one client create owners until others are evicted. +ObjOwner = Callable[[Any], Optional[Any]] + + +class OwnedObjStore: + """Objects kept per owner, with object and byte budgets. Thread-safe. + + - ``quota`` objects per owner: an owner's own least recently used object is evicted + beyond it, never someone else's; + - ``max_owners`` owners at once: the least recently active owner's objects go; + - byte accounting: :meth:`record` the size of an object; :meth:`owner_bytes` and + :attr:`total_bytes` sum what is currently stored. + + >>> s = OwnedObjStore(quota=2, max_owners=2) + >>> s.put('ann', 'a1', 1); s.put('ann', 'a2', 2); s.put('bob', 'b1', 3) + >>> s.put('ann', 'a3', 4) # ann's quota: her least recently used object goes + >>> s.get('ann', 'a1') is None, s.get('ann', 'a3'), s.get('bob', 'b1') + (True, 4, 3) + >>> s.get('bob', 'a3') is None # another owner's id is simply not found + True + >>> s.record('ann', 'a3', 100); s.record('bob', 'b1', 5) + >>> s.owner_bytes('ann'), s.total_bytes + (100, 105) + """ + + def __init__( + self, *, quota: int = DFLT_OBJ_QUOTA, max_owners: int = DFLT_MAX_OWNERS + ): + _at_least_one("max_owners", max_owners) + self._quota = _at_least_one("quota", quota) + self._lock = threading.RLock() + self._sizes: dict = {} # (owner, obj_id) -> bytes last recorded + self.total_bytes = 0 + self._owners = LRUObjStore(max_size=max_owners, on_evict=self._forget_owner) + + def _forget(self, owner, obj_id): + self.total_bytes -= self._sizes.pop((owner, obj_id), 0) + + def _forget_owner(self, owner, objects): + for obj_id in list(objects): + self._forget(owner, obj_id) + + def _objects_of(self, owner, *, create: bool): + objects = self._owners.get(owner) + if objects is None and create: + objects = LRUObjStore( + max_size=self._quota, + on_evict=lambda obj_id, _obj: self._forget(owner, obj_id), + ) + self._owners[owner] = objects + return objects + + def put(self, owner, obj_id: str, obj) -> None: + with self._lock: + self._objects_of(owner, create=True)[obj_id] = obj + + def get(self, owner, obj_id: str): + with self._lock: + objects = self._objects_of(owner, create=False) + return None if objects is None else objects.get(obj_id) + + def drop(self, owner, obj_id: str) -> None: + with self._lock: + objects = self._objects_of(owner, create=False) + if objects is not None: + objects.pop(obj_id, None) + self._forget(owner, obj_id) + + def record(self, owner, obj_id: str, size: int) -> None: + with self._lock: + if self.get(owner, obj_id) is None: + return # evicted meanwhile + self.total_bytes += size - self._sizes.get((owner, obj_id), 0) + self._sizes[(owner, obj_id)] = size + + def owner_bytes(self, owner) -> int: + with self._lock: + objects = self._owners.peek(owner) or {} + return sum(self._sizes.get((owner, obj_id), 0) for obj_id in objects) + + +def public_attr(obj, attr_name: str) -> bool: + """The attribute policy ``attr_names='*'`` stands for: only public methods. + + A name is dispatchable if it does not start with ``_`` and + + - for a module: it is a function that the module defines (or lists in its + ``__all__``), never something it merely imports, nor plain data; + - otherwise: it is a plain Python function, ``staticmethod``, + ``classmethod`` or ``property`` defined in the object's class (or a base + class), and the instance does not shadow it. + + This keeps private helpers, dunders (``__dict__``, ``__init__``, ...), + per-instance state (``__dict__`` entries, ``__slots__``, dataclass and + namedtuple fields) and callables merely assigned as class attributes + (``opener = open``) out of reach of HTTP clients. + + Note that every public method is exposed, inherited ones included (e.g. a + pydantic model's ``model_dump``). For anything holding secrets, or a mapping + whose mutators should stay private (``MutableMapping`` mixins such as + ``clear`` and ``pop`` are plain functions), list the names explicitly instead. + + >>> class A: + ... data = 0 + ... opener = open + ... def f(self): ... + ... def _g(self): ... + ... def __init__(self): self.secret = 1 + >>> [public_attr(A(), name) + ... for name in ('f', '_g', '__init__', '__dict__', 'secret', 'data', 'opener')] + [True, False, False, False, False, False, False] + """ + if not isinstance(attr_name, str) or attr_name.startswith("_"): + return False + if isinstance(obj, ModuleType): + return _is_public_module_function(obj, attr_name) + try: # (bypassing any class-level ``__dict__`` attribute) + instance_dict = object.__getattribute__(obj, "__dict__") + except AttributeError: + instance_dict = {} + if attr_name in instance_dict: + return False # instance state, or an instance attribute shadowing a method + # Look in the class's MRO dicts only (not the metaclass: ``mro``, ...), and at + # the raw class attribute, not what the instance would return for it + for klass in type(obj).__mro__: + if attr_name in vars(klass): + return _is_method_like(vars(klass)[attr_name]) + return False + + +def _is_method_like(class_attr) -> bool: + if isinstance(class_attr, (staticmethod, classmethod)): + class_attr = class_attr.__func__ + return isinstance(class_attr, (FunctionType, property)) + + +def _is_public_module_function(module: ModuleType, attr_name: str) -> bool: + value = vars(module).get(attr_name) + if not inspect.isfunction(value): + return False + exported = getattr(module, "__all__", None) + if exported is not None: + return attr_name in exported + return value.__module__ == module.__name__ + + +def unsafe_any_attr(obj, attr_name: str) -> bool: + """Dispatch ANY attribute of the object, including private ones and dunders. + + This was the behaviour of ``attr_names='*'`` before ``public_attr``: it lets a + client read ``__dict__``, call ``__init__`` with arguments of its choosing, and so + on. Use ``attr_names=unsafe_any_attr`` only when every client is trusted. + """ + return isinstance(attr_name, str) and hasattr(obj, attr_name) + + +def _attr_name_policy(attr_names) -> Callable[[Any, str], bool]: + """Normalize a handler's ``attr_names`` into a ``(obj, attr_name) -> bool`` + predicate: ``'*'`` means ``public_attr``, a callable is used as is, and an + iterable of names is an explicit allow-list (dunders included, if listed). + """ + if attr_names == "*": + return public_attr + if callable(attr_names): + return attr_names + if isinstance(attr_names, str): + allowed = {attr_names} + else: + allowed = set(attr_names) + return lambda obj, attr_name: ( + isinstance(attr_name, str) and attr_name in allowed and hasattr(obj, attr_name) + ) + + +def _get_func_to_dispatch_handler(handler): + endpoint = handler.get("endpoint") + if endpoint is None: + raise InputError("No endpoint found in handler") + name = handler.get("name", name_of_obj(endpoint)) + if not name: + raise InputError("No way to determine name of handler") + attr_names = handler.get("attr_names") + if attr_names is None: + # TODO: This is a hack! Address the problem in a cleaner way. + # This hack is just completing some unclean handling of names already present + # in the code. At the end of this function there's a func.__name__ = name, + # but this operation was forgotten here. That said, we shouldn't do this at + # all. + # --> When a function is specified for a handler, the long language specs + # needs to be created from it, and subsequently, the "name" field, not the + # __name__ attribute, needs to be used. + endpoint.__name__ = name # the hack + return endpoint + # was return endpoint, but then the output didn't get the __name__ = name + # func = endpoint + is_dispatchable = _attr_name_policy(attr_names) + if inspect.isclass(endpoint): + func = _class_endpoint_dispatcher(endpoint, handler, name, is_dispatchable) + + elif isinstance(endpoint, ModuleType): # a module's functions: no shared object + + def func(_attr_name=None, **kwargs): + return _get_attr_value(endpoint, _attr_name, is_dispatchable, **kwargs) + + else: + func = _instance_endpoint_dispatcher(endpoint, handler, name, is_dispatchable) + + func.__name__ = name + return func + + +class _MappingObjStore: + """``OwnedObjStore``'s interface over a user-given ``obj_store`` mapping + (``shared_objects`` mode): one owner, no byte accounting, a lock around access.""" + + total_bytes = 0 + + def __init__(self, mapping: MutableMapping): + self._mapping, self._lock = mapping, threading.RLock() + + def put(self, owner, obj_id, obj): + with self._lock: + self._mapping[obj_id] = obj + + def get(self, owner, obj_id): + with self._lock: + return self._mapping.get(obj_id) + + def drop(self, owner, obj_id): + with self._lock: + self._mapping.pop(obj_id, None) + + def record(self, owner, obj_id, size): + pass + + def owner_bytes(self, owner): + return 0 + + +_SHARED_OWNER = "" + + +def _request_owner(owner_of: ObjOwner, name: str) -> Callable[[], Any]: + """A zero-argument function giving the owner of the request being handled, as + ``owner_of(current_request)`` says, refusing requests it does not identify.""" + if not callable(owner_of): + raise TypeError(f"obj_owner must be callable or shared_objects: {owner_of!r}") + + def owner(): + who = owner_of(current_request.get()) + if who is None or who == "": + raise AuthorizationError(f"{name}: no owner for this request") + try: + hash(who) + except TypeError: + raise TypeError( + f"obj_owner of {name!r} must return a hashable, got {type(who)}" + ) from None + return who + + return owner + + +def _instance_endpoint_dispatcher(obj, handler, name, is_dispatchable): + """The function behind an instance endpoint (``handler['endpoint']`` is an object, + not a class or module): it dispatches ``_attr_name`` on that one object, which every + client allowed in shares. A method that mutates it (``__setitem__``, ``write``...) + lets each of them change what the others see. + + Handler options: + + - ``obj_owner`` (required): ``(request) -> owner`` (see ``ObjOwner``); a request it + does not identify is refused, the others share the object. Or ``shared_objects`` + to let any client use it. An instance endpoint without one is refused at build + time. + - ``max_obj_size`` (``DFLT_MAX_OBJ_SIZE``) and ``obj_sizeof`` + (``bounded_deep_sizeof``): the bytes the object may hold, measured when the + service is built (an object already over it is refused then) and after every + dispatched call, failed ones included. The call that takes it over the limit has + already run: its result is withheld, the object is kept, and every later call is + refused until it is back under, except calls to ``obj_shrinkers``. So memory is + bounded by the limit plus what the calls running at that moment allocate. + ``None`` disables the check. Raise ``max_obj_size``, or pass an ``obj_sizeof`` + that skips what is not request-driven (a loaded model, a database connection), + for a large object. Not measured: data the object keeps outside the process + (files, a database), and state held in its class or in module globals. + Measuring walks the object (up to ``max_obj_size`` bytes of it) after every + call, reads included, so a large limit costs time on every request. A method + returning a generator or coroutine does its work after the measure: what it + adds is caught by the next call. + - ``obj_shrinkers``: names of dispatched attributes that only ever remove + (``pop``, ``clear``, ``__delitem__`` if ``attr_names`` lists it...), still + callable while the object is over its limit so a client can bring it back under. + Empty by default: py2http cannot tell which methods shrink an object, and the + bound holds only if every name listed really only removes (one that also adds is + never refused). + """ + owner_of = handler.get("obj_owner") + if owner_of is None: + raise ValueError( + f"Instance endpoint {name!r} is one object shared by every client: give it " + "an obj_owner, a callable (request) -> owner (e.g. the authenticated user " + "id) that refuses requests it does not identify, or obj_owner=shared_objects " + "(from py2http.service) to let any client use it, as before." + ) + owner = ( + (lambda: None) if owner_of is shared_objects else _request_owner(owner_of, name) + ) + max_obj_size = handler.get("max_obj_size", DFLT_MAX_OBJ_SIZE) + sizeof = handler.get("obj_sizeof", bounded_deep_sizeof) + + def is_too_big(): + return max_obj_size is not None and sizeof(obj, max_obj_size) > max_obj_size + + if is_too_big(): + raise ValueError( + f"Instance endpoint {name!r} already holds more than max_obj_size=" + f"{max_obj_size} bytes (as measured by obj_sizeof): raise max_obj_size, pass " + "an obj_sizeof that skips what clients cannot grow, or max_obj_size=None." + ) + shrinkers = handler.get("obj_shrinkers", ()) + if isinstance(shrinkers, str): + raise TypeError(f"obj_shrinkers must be a collection of names: {shrinkers!r}") + shrinkers = frozenset(shrinkers) + for attr_name in shrinkers: + if not is_dispatchable(obj, attr_name): + raise ValueError( + f"obj_shrinkers of {name!r}: {attr_name!r} is not dispatched " + "(see attr_names)" + ) + over_limit = threading.Event() # set while the object is known to be too big + measuring = threading.Lock() # so a stale measure cannot clear a newer one's flag + + def measure() -> bool: + """Whether the object is over its limit, recorded in ``over_limit``.""" + with measuring: + too_big = is_too_big() + if too_big: + over_limit.set() + else: + over_limit.clear() + return too_big + + def refuse(): + allowed = f" (except {', '.join(sorted(shrinkers))})" if shrinkers else "" + raise InputError( + f"{name} is over its size limit ({max_obj_size} bytes): calls{allowed} " + "are refused until it is back under" + ) + + def func(_attr_name=None, **kwargs): + owner() + shrinking = isinstance(_attr_name, str) and _attr_name in shrinkers + if over_limit.is_set() and measure() and not shrinking: + refuse() + try: + return _get_attr_value(obj, _attr_name, is_dispatchable, **kwargs) + finally: # a call that grows the object and then raises is measured too + # a shrinker that ran but left it over the limit keeps its result + if measure() and not shrinking: + refuse() + + return func + + +def _class_endpoint_dispatcher(cls, handler, name, is_dispatchable): + """The function behind a class endpoint: called without ``_obj_id`` it creates an + object and returns its id; with one, it dispatches ``_attr_name`` on that object. + + Handler options: + + - ``obj_owner`` (required): ``(request) -> owner`` (see ``ObjOwner``), binding each + object to the client that created it (another owner's id is simply not found), or + ``shared_objects`` to let anyone holding an id use it. A class endpoint without + one is refused at build time. + - ``obj_quota``: objects kept per owner (``DFLT_OBJ_QUOTA``); an owner's least + recently used object is evicted beyond it. ``max_owners`` bounds the owners. + - ``obj_store``: ``shared_objects`` mode only, an id -> object mapping replacing the + default store (``DFLT_OBJ_STORE_MAX_SIZE`` objects); byte budgets other than + ``max_obj_size`` are then not tracked. + - ``max_obj_size`` (``DFLT_MAX_OBJ_SIZE``), ``max_owner_bytes`` + (``DFLT_MAX_OWNER_BYTES``) and ``max_total_bytes`` (``DFLT_MAX_TOTAL_BYTES``): + bytes one object, all of one owner's objects, and all objects of the endpoint may + hold. Sizes come from ``obj_sizeof`` (default ``bounded_deep_sizeof``, which + stops at ``max_obj_size``, so measuring costs at most about that much), measured + after creation and after every dispatched call, failed ones included. The object + that takes a budget over it is dropped. ``None`` disables a budget. An object + referencing large shared data (a loaded model) counts that data: raise + ``max_obj_size`` or pass an ``obj_sizeof`` that skips it. A single call can still + allocate before it is measured. + """ + owner_of = handler.get("obj_owner") + if owner_of is None: + raise ValueError( + f"Class endpoint {name!r} needs an obj_owner: a callable (request) -> owner " + "(e.g. the authenticated user id), so a client can only use the objects it " + "created. Pass obj_owner=shared_objects (from py2http.service) to let any " + "client holding an object id use the object, as before." + ) + max_obj_size = handler.get("max_obj_size", DFLT_MAX_OBJ_SIZE) + max_owner_bytes = handler.get("max_owner_bytes", DFLT_MAX_OWNER_BYTES) + max_total_bytes = handler.get("max_total_bytes", DFLT_MAX_TOTAL_BYTES) + sizeof = handler.get("obj_sizeof", bounded_deep_sizeof) + + if owner_of is shared_objects: + custom = handler.get("obj_store") + store = ( + OwnedObjStore(quota=DFLT_OBJ_STORE_MAX_SIZE, max_owners=1) + if custom is None + else _MappingObjStore(custom) + ) + + def owner(): + return _SHARED_OWNER + + else: + owner = _request_owner(owner_of, name) + if handler.get("obj_store") is not None: + raise ValueError( + "obj_store applies to obj_owner=shared_objects only; with an obj_owner " + "use obj_quota and max_owners." + ) + store = OwnedObjStore( + quota=handler.get("obj_quota", DFLT_OBJ_QUOTA), + max_owners=handler.get("max_owners", DFLT_MAX_OWNERS), + ) + + def over_budget(who, obj_id, obj): + """Why the object must go, or None. Records its size when it may stay.""" + size = sizeof(obj, max_obj_size) + if max_obj_size is not None and size > max_obj_size: + return f"grew past the size limit ({max_obj_size} bytes)" + store.record(who, obj_id, size) + if max_owner_bytes is not None and store.owner_bytes(who) > max_owner_bytes: + return f"takes its owner past {max_owner_bytes} bytes" + if max_total_bytes is not None and store.total_bytes > max_total_bytes: + return f"takes the endpoint past {max_total_bytes} bytes" + return None + + def check_size(who, obj_id, obj): + no_budget = max_obj_size is max_owner_bytes is max_total_bytes is None + reason = None if no_budget else over_budget(who, obj_id, obj) + if reason is not None: + store.drop(who, obj_id) + raise InputError(f"Object {obj_id} {reason} and was dropped") + + def func(_obj_id=None, _attr_name=None, **kwargs): + who = owner() + if _obj_id is None: + if _attr_name is not None: + raise InputError("_attr_name must be None when _obj_id is None") + new_obj = call_with_inputs(cls, kwargs=kwargs) + new_id = str(uuid4()) + store.put(who, new_id, new_obj) + check_size(who, new_id, new_obj) + return new_id + obj = store.get(who, _obj_id) if isinstance(_obj_id, str) else None + if obj is None: + raise InputError(f"No object found with id {_obj_id}") + if _attr_name is None: + raise InputError("_attr_name must be provided when _obj_id is not None") + try: + return _get_attr_value(obj, _attr_name, is_dispatchable, **kwargs) + finally: # a call that grows the object and then raises is measured too + check_size(who, _obj_id, obj) + + return func + + +def _get_attr_value(obj, attr_name, is_dispatchable, /, **kwargs): + if not is_dispatchable(obj, attr_name): + raise InputError( + f"No attribute found with name {attr_name} or it is not dispatched" + ) + attr = getattr(obj, attr_name) + if callable(attr): + return call_with_inputs(attr, kwargs=kwargs) + return attr diff --git a/py2http/service.py b/py2http/service.py index a70d724..56ecf7b 100644 --- a/py2http/service.py +++ b/py2http/service.py @@ -15,24 +15,16 @@ Python functions, enabling developers to easily expose their functions as remote APIs with minimal setup.""" -import inspect -from collections import OrderedDict, deque -from contextvars import ContextVar -from collections.abc import MutableMapping -from types import ModuleType -from uuid import uuid4 from aiohttp import web import asyncio from copy import deepcopy -from functools import partial, wraps +from functools import partial from inspect import isawaitable import json from typing import Any, Callable, Dict, Iterable, Optional, TypedDict, Union -from types import BuiltinMethodType, FunctionType, MethodType, MethodWrapperType +from types import FunctionType import logging import os -import sys -import threading from bottle import Bottle, run as run_bottle import traceback from swagger_ui import api_doc @@ -59,353 +51,25 @@ from py2http.util import TypeAsserter, call_with_inputs from py2http.constants import JSON_CONTENT_TYPE -# Instance endpoints (``handler['endpoint']`` is an object) are one object every client -# shares; they too need an ``obj_owner`` and are bounded by ``max_obj_size`` (see -# ``_instance_endpoint_dispatcher``). -# -# Class endpoints (``handler['endpoint']`` is a class) keep the objects clients create. -# Who may use an object is the handler's ``obj_owner`` (see ``ObjOwner``); how many -# objects an owner may keep is ``obj_quota``; how many bytes each object, each owner and -# the whole endpoint may hold are ``max_obj_size``, ``max_owner_bytes`` and -# ``max_total_bytes`` (see ``_class_endpoint_dispatcher``). -# -# Maximum number of objects a class endpoint keeps alive in ``shared_objects`` mode; the -# least recently used one is evicted beyond that. Pass ``obj_store=...`` to change it. -DFLT_OBJ_STORE_MAX_SIZE = 1000 -DFLT_OBJ_QUOTA = 100 # objects per owner -DFLT_MAX_OWNERS = 1000 # owners holding objects at once (least recently active evicted) -DFLT_MAX_OBJ_SIZE = 1_000_000 # bytes per object, as measured by ``obj_sizeof`` -DFLT_MAX_OWNER_BYTES = 20_000_000 # bytes per owner (all their objects) -DFLT_MAX_TOTAL_BYTES = 256_000_000 # bytes per endpoint (every object of every owner) - -#: The request being handled (set around every handler call), so per-request hooks such -#: as ``obj_owner`` can read it without the dispatched function taking it. -current_request: ContextVar = ContextVar("py2http_current_request", default=None) - - -class LRUObjStore(MutableMapping): - """A mapping that keeps at most ``max_size`` items, evicting the least recently - used one when a new key would exceed it (``on_evict(k, v)`` is told). - - Used as the default store of objects created through a class endpoint, so that - clients cannot grow server memory without bound by creating objects. - - >>> s = LRUObjStore(max_size=2) - >>> s['a'], s['b'] = 1, 2 - >>> _ = s['a'] # 'a' is now the most recently used - >>> s['c'] = 3 - >>> sorted(s) - ['a', 'c'] - """ - - def __init__( - self, - *, - max_size: int = DFLT_OBJ_STORE_MAX_SIZE, - on_evict: Optional[Callable[[Any, Any], None]] = None, - ): - if max_size < 1: - raise ValueError(f"max_size must be at least 1, got {max_size}") - self.max_size = max_size - self.on_evict = on_evict - self._data = OrderedDict() - - def __getitem__(self, k): - v = self._data[k] - self._data.move_to_end(k) - return v - - def __setitem__(self, k, v): - self._data[k] = v - self._data.move_to_end(k) - while len(self._data) > self.max_size: - evicted_k, evicted_v = self._data.popitem(last=False) - if self.on_evict is not None: - self.on_evict(evicted_k, evicted_v) - - def __delitem__(self, k): - del self._data[k] - - def __iter__(self): - return iter(self._data) - - def __len__(self): - return len(self._data) - - -class _SharedObjects: - """``obj_owner=shared_objects``: objects belong to no one. Any client holding an - object id may use it, and every client's objects share one LRU store, so a client - creating objects evicts everyone else's. Only for trusted clients.""" - - def __repr__(self): - return "shared_objects" - - -shared_objects = _SharedObjects() - -#: ``obj_owner`` hook: ``(request) -> owner``, a hashable that identifies an -#: *authenticated* client (e.g. a user id from a verified token). ``None`` and ``""`` -#: mean "not identified" and are refused. A client-chosen value (a session cookie the -#: client mints, an IP) lets one client create owners until others are evicted. -ObjOwner = Callable[[Any], Optional[Any]] - - -class OwnedObjStore: - """Objects kept per owner, with object and byte budgets. Thread-safe. - - - ``quota`` objects per owner: an owner's own least recently used object is evicted - beyond it, never someone else's; - - ``max_owners`` owners at once: the least recently active owner's objects go; - - byte accounting: :meth:`record` the size of an object; :meth:`owner_bytes` and - :attr:`total_bytes` sum what is currently stored. - - >>> s = OwnedObjStore(quota=2, max_owners=2) - >>> s.put('ann', 'a1', 1); s.put('ann', 'a2', 2); s.put('bob', 'b1', 3) - >>> s.put('ann', 'a3', 4) # ann's quota: her least recently used object goes - >>> s.get('ann', 'a1') is None, s.get('ann', 'a3'), s.get('bob', 'b1') - (True, 4, 3) - >>> s.get('bob', 'a3') is None # another owner's id is simply not found - True - >>> s.record('ann', 'a3', 100); s.record('bob', 'b1', 5) - >>> s.owner_bytes('ann'), s.total_bytes - (100, 105) - """ - - def __init__( - self, *, quota: int = DFLT_OBJ_QUOTA, max_owners: int = DFLT_MAX_OWNERS - ): - if max_owners < 1: - raise ValueError(f"max_owners must be at least 1, got {max_owners}") - LRUObjStore(max_size=quota) # validates quota now, not on first use - self._quota = quota - self._lock = threading.RLock() - self._sizes: dict = {} # (owner, obj_id) -> bytes last recorded - self.total_bytes = 0 - self._owners = LRUObjStore(max_size=max_owners, on_evict=self._forget_owner) - - def _forget(self, owner, obj_id): - self.total_bytes -= self._sizes.pop((owner, obj_id), 0) - - def _forget_owner(self, owner, objects): - for obj_id in list(objects): - self._forget(owner, obj_id) - - def _objects_of(self, owner, *, create: bool): - objects = self._owners.get(owner) - if objects is None and create: - objects = LRUObjStore( - max_size=self._quota, - on_evict=lambda obj_id, _obj: self._forget(owner, obj_id), - ) - self._owners[owner] = objects - return objects - - def put(self, owner, obj_id: str, obj) -> None: - with self._lock: - self._objects_of(owner, create=True)[obj_id] = obj - - def get(self, owner, obj_id: str): - with self._lock: - objects = self._objects_of(owner, create=False) - return None if objects is None else objects.get(obj_id) - - def drop(self, owner, obj_id: str) -> None: - with self._lock: - objects = self._objects_of(owner, create=False) - if objects is not None: - objects.pop(obj_id, None) - self._forget(owner, obj_id) - - def record(self, owner, obj_id: str, size: int) -> None: - with self._lock: - if self.get(owner, obj_id) is None: - return # evicted meanwhile - self.total_bytes += size - self._sizes.get((owner, obj_id), 0) - self._sizes[(owner, obj_id)] = size - - def owner_bytes(self, owner) -> int: - with self._lock: - objects = self._owners._data.get(owner) or {} - return sum(self._sizes.get((owner, obj_id), 0) for obj_id in objects) - - -_SHARED_SIZE_LEAVES = (type, ModuleType) -_BOUND_METHODS = (MethodType, BuiltinMethodType, MethodWrapperType) - - -def bounded_deep_sizeof(obj, limit: Optional[int] = None) -> int: - """Approximate memory held by ``obj`` and everything it references (containers, - instance ``__dict__`` and ``__slots__``, what a bound method, a ``partial`` or a - function's closure, attributes and defaults hold), counting each object once. Stops - as soon as the total exceeds ``limit``, so measuring costs at most about ``limit`` - bytes' worth of objects. Classes, modules and the code of functions are shared, so - not counted, and neither is what classes and modules hold (class attributes, module - globals), nor caches kept in C (``lru_cache``). Holding a bound method of another - object counts that whole object, as holding the object itself would. - - >>> small, big = bounded_deep_sizeof([1, 2]), bounded_deep_sizeof(list(range(1000))) - >>> small < big - True - >>> bounded_deep_sizeof(list(range(10**6)), limit=1000) > 1000 - True - """ - seen, total, todo = set(), 0, deque([obj]) - while todo: - o = todo.pop() - if id(o) in seen or isinstance(o, _SHARED_SIZE_LEAVES): - continue - seen.add(id(o)) - try: - held = _held_by_callable(o) - except Exception: # an exotic object (a mock, a proxy): measure it plainly - held = None - if isinstance(o, FunctionType): # the code is shared; only what it holds counts - todo.extend(held or ()) - continue - todo.extend(held or ()) - total += sys.getsizeof(o, 0) - if limit is not None and total > limit: - break - if isinstance(o, dict): - todo.extend(o.keys()) - todo.extend(o.values()) - elif isinstance(o, (list, tuple, set, frozenset, deque)): - todo.extend(o) - try: # a __getattr__ may raise anything - todo.extend(_instance_state(o)) - except Exception: - pass - return total - - -def _instance_state(o): - """An object's ``__dict__`` and the values of its ``__slots__``.""" - if hasattr(o, "__dict__") and isinstance(vars(o), dict): - yield vars(o) - for klass in type(o).__mro__: - slots = getattr(klass, "__slots__", ()) - for slot in (slots,) if isinstance(slots, str) else slots: - if slot not in ("__dict__", "__weakref__") and hasattr(o, slot): - yield getattr(o, slot) - - -def _held_by_callable(o) -> Optional[list]: - """What a function (closure, attributes, defaults), bound method, static or class - method, or ``partial`` holds; ``None`` for anything else.""" - if isinstance(o, FunctionType): - cells = [c.cell_contents for c in o.__closure__ or () if _cell_is_set(c)] - return [*cells, o.__dict__, o.__defaults__, o.__kwdefaults__] - if isinstance(o, MethodType): - return [o.__self__, o.__func__] - if isinstance(o, _BOUND_METHODS): # a builtin's: what it is bound to - return [o.__self__] - if isinstance(o, (staticmethod, classmethod)): - return [o.__func__] - if isinstance(o, partial): - return [o.func, o.args, o.keywords] - return None - - -def _cell_is_set(cell) -> bool: - try: - cell.cell_contents - except ValueError: # an empty cell (a free variable not yet assigned) - return False - return True - - -def public_attr(obj, attr_name: str) -> bool: - """The attribute policy ``attr_names='*'`` stands for: only public methods. - - A name is dispatchable if it does not start with ``_`` and - - - for a module: it is a function that the module defines (or lists in its - ``__all__``), never something it merely imports, nor plain data; - - otherwise: it is a plain Python function, ``staticmethod``, - ``classmethod`` or ``property`` defined in the object's class (or a base - class), and the instance does not shadow it. - - This keeps private helpers, dunders (``__dict__``, ``__init__``, ...), - per-instance state (``__dict__`` entries, ``__slots__``, dataclass and - namedtuple fields) and callables merely assigned as class attributes - (``opener = open``) out of reach of HTTP clients. - - Note that every public method is exposed, inherited ones included (e.g. a - pydantic model's ``model_dump``). For anything holding secrets, or a mapping - whose mutators should stay private (``MutableMapping`` mixins such as - ``clear`` and ``pop`` are plain functions), list the names explicitly instead. - - >>> class A: - ... data = 0 - ... opener = open - ... def f(self): ... - ... def _g(self): ... - ... def __init__(self): self.secret = 1 - >>> [public_attr(A(), name) - ... for name in ('f', '_g', '__init__', '__dict__', 'secret', 'data', 'opener')] - [True, False, False, False, False, False, False] - """ - if not isinstance(attr_name, str) or attr_name.startswith("_"): - return False - if isinstance(obj, ModuleType): - return _is_public_module_function(obj, attr_name) - try: # (bypassing any class-level ``__dict__`` attribute) - instance_dict = object.__getattribute__(obj, "__dict__") - except AttributeError: - instance_dict = {} - if attr_name in instance_dict: - return False # instance state, or an instance attribute shadowing a method - # Look in the class's MRO dicts only (not the metaclass: ``mro``, ...), and at - # the raw class attribute, not what the instance would return for it - for klass in type(obj).__mro__: - if attr_name in vars(klass): - return _is_method_like(vars(klass)[attr_name]) - return False - - -def _is_method_like(class_attr) -> bool: - if isinstance(class_attr, (staticmethod, classmethod)): - class_attr = class_attr.__func__ - return isinstance(class_attr, (FunctionType, property)) - - -def _is_public_module_function(module: ModuleType, attr_name: str) -> bool: - value = vars(module).get(attr_name) - if not inspect.isfunction(value): - return False - exported = getattr(module, "__all__", None) - if exported is not None: - return attr_name in exported - return value.__module__ == module.__name__ - - -def unsafe_any_attr(obj, attr_name: str) -> bool: - """Dispatch ANY attribute of the object, including private ones and dunders. - - This was the behaviour of ``attr_names='*'`` before ``public_attr``: it lets a - client read ``__dict__``, call ``__init__`` with arguments of its choosing, and so - on. Use ``attr_names=unsafe_any_attr`` only when every client is trusted. - """ - return isinstance(attr_name, str) and hasattr(obj, attr_name) - - -def _attr_name_policy(attr_names) -> Callable[[Any, str], bool]: - """Normalize a handler's ``attr_names`` into a ``(obj, attr_name) -> bool`` - predicate: ``'*'`` means ``public_attr``, a callable is used as is, and an - iterable of names is an explicit allow-list (dunders included, if listed). - """ - if attr_names == "*": - return public_attr - if callable(attr_names): - return attr_names - if isinstance(attr_names, str): - allowed = {attr_names} - else: - allowed = set(attr_names) - return lambda obj, attr_name: ( - isinstance(attr_name, str) and attr_name in allowed and hasattr(obj, attr_name) - ) +# Object dispatch moved to py2http.obj_dispatch (and py2http.sizeof); these names stay +# importable from here. +from py2http.obj_dispatch import ( # noqa: F401 + DFLT_MAX_OBJ_SIZE, + DFLT_MAX_OWNER_BYTES, + DFLT_MAX_OWNERS, + DFLT_MAX_TOTAL_BYTES, + DFLT_OBJ_QUOTA, + DFLT_OBJ_STORE_MAX_SIZE, + LRUObjStore, + ObjOwner, + OwnedObjStore, + _get_func_to_dispatch_handler, + current_request, + public_attr, + shared_objects, + unsafe_any_attr, +) +from py2http.sizeof import bounded_deep_sizeof # noqa: F401 def method_not_found(method_name): @@ -957,310 +621,3 @@ def _get_framework(configs, default_configs): ) os.environ["PY2HTTP_FRAMEWORK"] = framework return framework - - -def _get_func_to_dispatch_handler(handler): - endpoint = handler.get("endpoint") - if endpoint is None: - raise InputError("No endpoint found in handler") - name = handler.get("name", name_of_obj(endpoint)) - if not name: - raise InputError("No way to determine name of handler") - attr_names = handler.get("attr_names") - if attr_names is None: - # TODO: This is a hack! Address the problem in a cleaner way. - # This hack is just completing some unclean handling of names already present - # in the code. At the end of this function there's a func.__name__ = name, - # but this operation was forgotten here. That said, we shouldn't do this at - # all. - # --> When a function is specified for a handler, the long language specs - # needs to be created from it, and subsequently, the "name" field, not the - # __name__ attribute, needs to be used. - endpoint.__name__ = name # the hack - return endpoint - # was return endpoint, but then the output didn't get the __name__ = name - # func = endpoint - is_dispatchable = _attr_name_policy(attr_names) - if inspect.isclass(endpoint): - func = _class_endpoint_dispatcher(endpoint, handler, name, is_dispatchable) - - elif isinstance(endpoint, ModuleType): # a module's functions: no shared object - - def func(_attr_name=None, **kwargs): - return _get_attr_value(endpoint, _attr_name, is_dispatchable, **kwargs) - - else: - func = _instance_endpoint_dispatcher(endpoint, handler, name, is_dispatchable) - - func.__name__ = name - return func - - -class _MappingObjStore: - """``OwnedObjStore``'s interface over a user-given ``obj_store`` mapping - (``shared_objects`` mode): one owner, no byte accounting, a lock around access.""" - - total_bytes = 0 - - def __init__(self, mapping: MutableMapping): - self._mapping, self._lock = mapping, threading.RLock() - - def put(self, owner, obj_id, obj): - with self._lock: - self._mapping[obj_id] = obj - - def get(self, owner, obj_id): - with self._lock: - return self._mapping.get(obj_id) - - def drop(self, owner, obj_id): - with self._lock: - self._mapping.pop(obj_id, None) - - def record(self, owner, obj_id, size): - pass - - def owner_bytes(self, owner): - return 0 - - -_SHARED_OWNER = "" - - -def _request_owner(owner_of: ObjOwner, name: str) -> Callable[[], Any]: - """A zero-argument function giving the owner of the request being handled, as - ``owner_of(current_request)`` says, refusing requests it does not identify.""" - if not callable(owner_of): - raise TypeError(f"obj_owner must be callable or shared_objects: {owner_of!r}") - - def owner(): - who = owner_of(current_request.get()) - if who is None or who == "": - raise AuthorizationError(f"{name}: no owner for this request") - try: - hash(who) - except TypeError: - raise TypeError( - f"obj_owner of {name!r} must return a hashable, got {type(who)}" - ) from None - return who - - return owner - - -def _instance_endpoint_dispatcher(obj, handler, name, is_dispatchable): - """The function behind an instance endpoint (``handler['endpoint']`` is an object, - not a class or module): it dispatches ``_attr_name`` on that one object, which every - client allowed in shares. A method that mutates it (``__setitem__``, ``write``...) - lets each of them change what the others see. - - Handler options: - - - ``obj_owner`` (required): ``(request) -> owner`` (see ``ObjOwner``); a request it - does not identify is refused, the others share the object. Or ``shared_objects`` - to let any client use it. An instance endpoint without one is refused at build - time. - - ``max_obj_size`` (``DFLT_MAX_OBJ_SIZE``) and ``obj_sizeof`` - (``bounded_deep_sizeof``): the bytes the object may hold, measured when the - service is built (an object already over it is refused then) and after every - dispatched call, failed ones included. The call that takes it over the limit has - already run: its result is withheld, the object is kept, and every later call is - refused until it is back under, except calls to ``obj_shrinkers``. So memory is - bounded by the limit plus what the calls running at that moment allocate. - ``None`` disables the check. Raise ``max_obj_size``, or pass an ``obj_sizeof`` - that skips what is not request-driven (a loaded model, a database connection), - for a large object. Not measured: data the object keeps outside the process - (files, a database), and state held in its class or in module globals. - Measuring walks the object (up to ``max_obj_size`` bytes of it) after every - call, reads included, so a large limit costs time on every request. A method - returning a generator or coroutine does its work after the measure: what it - adds is caught by the next call. - - ``obj_shrinkers``: names of dispatched attributes that only ever remove - (``pop``, ``clear``, ``__delitem__`` if ``attr_names`` lists it...), still - callable while the object is over its limit so a client can bring it back under. - Empty by default: py2http cannot tell which methods shrink an object, and the - bound holds only if every name listed really only removes (one that also adds is - never refused). - """ - owner_of = handler.get("obj_owner") - if owner_of is None: - raise ValueError( - f"Instance endpoint {name!r} is one object shared by every client: give it " - "an obj_owner, a callable (request) -> owner (e.g. the authenticated user " - "id) that refuses requests it does not identify, or obj_owner=shared_objects " - "(from py2http.service) to let any client use it, as before." - ) - owner = ( - (lambda: None) if owner_of is shared_objects else _request_owner(owner_of, name) - ) - max_obj_size = handler.get("max_obj_size", DFLT_MAX_OBJ_SIZE) - sizeof = handler.get("obj_sizeof", bounded_deep_sizeof) - - def is_too_big(): - return max_obj_size is not None and sizeof(obj, max_obj_size) > max_obj_size - - if is_too_big(): - raise ValueError( - f"Instance endpoint {name!r} already holds more than max_obj_size=" - f"{max_obj_size} bytes (as measured by obj_sizeof): raise max_obj_size, pass " - "an obj_sizeof that skips what clients cannot grow, or max_obj_size=None." - ) - shrinkers = handler.get("obj_shrinkers", ()) - if isinstance(shrinkers, str): - raise TypeError(f"obj_shrinkers must be a collection of names: {shrinkers!r}") - shrinkers = frozenset(shrinkers) - for attr_name in shrinkers: - if not is_dispatchable(obj, attr_name): - raise ValueError( - f"obj_shrinkers of {name!r}: {attr_name!r} is not dispatched " - "(see attr_names)" - ) - over_limit = threading.Event() # set while the object is known to be too big - measuring = threading.Lock() # so a stale measure cannot clear a newer one's flag - - def measure() -> bool: - """Whether the object is over its limit, recorded in ``over_limit``.""" - with measuring: - too_big = is_too_big() - if too_big: - over_limit.set() - else: - over_limit.clear() - return too_big - - def refuse(): - allowed = f" (except {', '.join(sorted(shrinkers))})" if shrinkers else "" - raise InputError( - f"{name} is over its size limit ({max_obj_size} bytes): calls{allowed} " - "are refused until it is back under" - ) - - def func(_attr_name=None, **kwargs): - owner() - shrinking = isinstance(_attr_name, str) and _attr_name in shrinkers - if over_limit.is_set() and measure() and not shrinking: - refuse() - try: - return _get_attr_value(obj, _attr_name, is_dispatchable, **kwargs) - finally: # a call that grows the object and then raises is measured too - # a shrinker that ran but left it over the limit keeps its result - if measure() and not shrinking: - refuse() - - return func - - -def _class_endpoint_dispatcher(cls, handler, name, is_dispatchable): - """The function behind a class endpoint: called without ``_obj_id`` it creates an - object and returns its id; with one, it dispatches ``_attr_name`` on that object. - - Handler options: - - - ``obj_owner`` (required): ``(request) -> owner`` (see ``ObjOwner``), binding each - object to the client that created it (another owner's id is simply not found), or - ``shared_objects`` to let anyone holding an id use it. A class endpoint without - one is refused at build time. - - ``obj_quota``: objects kept per owner (``DFLT_OBJ_QUOTA``); an owner's least - recently used object is evicted beyond it. ``max_owners`` bounds the owners. - - ``obj_store``: ``shared_objects`` mode only, an id -> object mapping replacing the - default store (``DFLT_OBJ_STORE_MAX_SIZE`` objects); byte budgets other than - ``max_obj_size`` are then not tracked. - - ``max_obj_size`` (``DFLT_MAX_OBJ_SIZE``), ``max_owner_bytes`` - (``DFLT_MAX_OWNER_BYTES``) and ``max_total_bytes`` (``DFLT_MAX_TOTAL_BYTES``): - bytes one object, all of one owner's objects, and all objects of the endpoint may - hold. Sizes come from ``obj_sizeof`` (default ``bounded_deep_sizeof``, which - stops at ``max_obj_size``, so measuring costs at most about that much), measured - after creation and after every dispatched call, failed ones included. The object - that takes a budget over it is dropped. ``None`` disables a budget. An object - referencing large shared data (a loaded model) counts that data: raise - ``max_obj_size`` or pass an ``obj_sizeof`` that skips it. A single call can still - allocate before it is measured. - """ - owner_of = handler.get("obj_owner") - if owner_of is None: - raise ValueError( - f"Class endpoint {name!r} needs an obj_owner: a callable (request) -> owner " - "(e.g. the authenticated user id), so a client can only use the objects it " - "created. Pass obj_owner=shared_objects (from py2http.service) to let any " - "client holding an object id use the object, as before." - ) - max_obj_size = handler.get("max_obj_size", DFLT_MAX_OBJ_SIZE) - max_owner_bytes = handler.get("max_owner_bytes", DFLT_MAX_OWNER_BYTES) - max_total_bytes = handler.get("max_total_bytes", DFLT_MAX_TOTAL_BYTES) - sizeof = handler.get("obj_sizeof", bounded_deep_sizeof) - - if owner_of is shared_objects: - custom = handler.get("obj_store") - store = ( - OwnedObjStore(quota=DFLT_OBJ_STORE_MAX_SIZE, max_owners=1) - if custom is None - else _MappingObjStore(custom) - ) - - def owner(): - return _SHARED_OWNER - - else: - owner = _request_owner(owner_of, name) - if handler.get("obj_store") is not None: - raise ValueError( - "obj_store applies to obj_owner=shared_objects only; with an obj_owner " - "use obj_quota and max_owners." - ) - store = OwnedObjStore( - quota=handler.get("obj_quota", DFLT_OBJ_QUOTA), - max_owners=handler.get("max_owners", DFLT_MAX_OWNERS), - ) - - def over_budget(who, obj_id, obj): - """Why the object must go, or None. Records its size when it may stay.""" - limit = max_obj_size - size = sizeof(obj, limit) if limit is not None else sizeof(obj, None) - if limit is not None and size > limit: - return f"grew past the size limit ({limit} bytes)" - store.record(who, obj_id, size) - if max_owner_bytes is not None and store.owner_bytes(who) > max_owner_bytes: - return f"takes its owner past {max_owner_bytes} bytes" - if max_total_bytes is not None and store.total_bytes > max_total_bytes: - return f"takes the endpoint past {max_total_bytes} bytes" - return None - - def check_size(who, obj_id, obj): - no_budget = max_obj_size is max_owner_bytes is max_total_bytes is None - reason = None if no_budget else over_budget(who, obj_id, obj) - if reason is not None: - store.drop(who, obj_id) - raise InputError(f"Object {obj_id} {reason} and was dropped") - - def func(_obj_id=None, _attr_name=None, **kwargs): - who = owner() - if _obj_id is None: - if _attr_name is not None: - raise InputError("_attr_name must be None when _obj_id is None") - new_obj = call_with_inputs(cls, kwargs=kwargs) - new_id = str(uuid4()) - store.put(who, new_id, new_obj) - check_size(who, new_id, new_obj) - return new_id - obj = store.get(who, _obj_id) if isinstance(_obj_id, str) else None - if obj is None: - raise InputError(f"No object found with id {_obj_id}") - if _attr_name is None: - raise InputError("_attr_name must be provided when _obj_id is not None") - try: - return _get_attr_value(obj, _attr_name, is_dispatchable, **kwargs) - finally: # a call that grows the object and then raises is measured too - check_size(who, _obj_id, obj) - - return func - - -def _get_attr_value(obj, attr_name, is_dispatchable, /, **kwargs): - if not is_dispatchable(obj, attr_name): - raise InputError( - f"No attribute found with name {attr_name} or it is not dispatched" - ) - attr = getattr(obj, attr_name) - if callable(attr): - return call_with_inputs(attr, kwargs=kwargs) - return attr diff --git a/py2http/sizeof.py b/py2http/sizeof.py new file mode 100644 index 0000000..bf5df81 --- /dev/null +++ b/py2http/sizeof.py @@ -0,0 +1,101 @@ +"""Approximate the memory an object holds, without walking more of it than a budget. + +``bounded_deep_sizeof`` is how py2http bounds the objects clients create or grow +through class and instance endpoints (``py2http.obj_dispatch``). +""" + +import sys +from collections import deque +from functools import partial +from types import ( + BuiltinMethodType, + FunctionType, + MethodType, + MethodWrapperType, + ModuleType, +) +from typing import Optional + +_SHARED_SIZE_LEAVES = (type, ModuleType) +_BOUND_METHODS = (MethodType, BuiltinMethodType, MethodWrapperType) + + +def bounded_deep_sizeof(obj, limit: Optional[int] = None) -> int: + """Approximate memory held by ``obj`` and everything it references (containers, + instance ``__dict__`` and ``__slots__``, what a bound method, a ``partial`` or a + function's closure, attributes and defaults hold), counting each object once. Stops + as soon as the total exceeds ``limit``, so measuring costs at most about ``limit`` + bytes' worth of objects. Classes, modules and the code of functions are shared, so + not counted, and neither is what classes and modules hold (class attributes, module + globals), nor caches kept in C (``lru_cache``). Holding a bound method of another + object counts that whole object, as holding the object itself would. + + >>> small, big = bounded_deep_sizeof([1, 2]), bounded_deep_sizeof(list(range(1000))) + >>> small < big + True + >>> bounded_deep_sizeof(list(range(10**6)), limit=1000) > 1000 + True + """ + seen, total, todo = set(), 0, deque([obj]) + while todo: + o = todo.pop() + if id(o) in seen or isinstance(o, _SHARED_SIZE_LEAVES): + continue + seen.add(id(o)) + try: + held = _held_by_callable(o) + except Exception: # an exotic object (a mock, a proxy): measure it plainly + held = None + if isinstance(o, FunctionType): # the code is shared; only what it holds counts + todo.extend(held or ()) + continue + todo.extend(held or ()) + total += sys.getsizeof(o, 0) + if limit is not None and total > limit: + break + if isinstance(o, dict): + todo.extend(o.keys()) + todo.extend(o.values()) + elif isinstance(o, (list, tuple, set, frozenset, deque)): + todo.extend(o) + try: # a __getattr__ may raise anything + todo.extend(_instance_state(o)) + except Exception: + pass + return total + + +def _instance_state(o): + """An object's ``__dict__`` and the values of its ``__slots__``.""" + if hasattr(o, "__dict__") and isinstance(vars(o), dict): + yield vars(o) + for klass in type(o).__mro__: + slots = getattr(klass, "__slots__", ()) + for slot in (slots,) if isinstance(slots, str) else slots: + if slot not in ("__dict__", "__weakref__") and hasattr(o, slot): + yield getattr(o, slot) + + +def _held_by_callable(o) -> Optional[list]: + """What a function (closure, attributes, defaults), bound method, static or class + method, or ``partial`` holds; ``None`` for anything else.""" + if isinstance(o, FunctionType): + cells = [c.cell_contents for c in o.__closure__ or () if _cell_is_set(c)] + return [*cells, o.__dict__, o.__defaults__, o.__kwdefaults__] + if isinstance(o, MethodType): + return [o.__self__, o.__func__] + if isinstance(o, _BOUND_METHODS): # a builtin's: what it is bound to + return [o.__self__] + if isinstance(o, (staticmethod, classmethod)): + return [o.__func__] + if isinstance(o, partial): + return [o.func, o.args, o.keywords] + return None + + +def _cell_is_set(cell) -> bool: + try: + cell.cell_contents + except ValueError: # an empty cell (a free variable not yet assigned) + return False + return True diff --git a/py2http/tests/test_auth_plugins.py b/py2http/tests/test_auth_plugins.py new file mode 100644 index 0000000..8923ee1 --- /dev/null +++ b/py2http/tests/test_auth_plugins.py @@ -0,0 +1,349 @@ +"""The Bottle auth plugins (``JWTPlugin``, ``ApiKeyAuthPlugin``) and the aiohttp +``mk_superadmin_middleware`` let valid credentials through and refuse the rest. + +Bottle is the framework ``mk_app`` builds, so ``JWTPlugin`` is what guards a py2http +service; these tests go through a real ``mk_app`` app. +""" + +import asyncio +import base64 +import json +import time +import warnings + +import jwt +import pytest +from aiohttp import web +from aiohttp.test_utils import TestClient, TestServer + +from py2http import mk_app +from py2http.bottle_plugins import ApiKeyAuthPlugin, JWTPlugin +from py2http.middleware import mk_jwt_middleware, mk_superadmin_middleware +from py2http.service import current_request +from py2http.tests.utils_for_testing import call_wsgi + +SECRET = "s3cret-for-tests-only-0123456789abcdef" +API_KEY = "api-key-for-tests-only" + + +def whoami(): + return getattr(current_request.get(), "token", None) + + +def other(): + return "other" + + +def _app(*plugins): + return mk_app([whoami, other], plugins=list(plugins)) + + +def _bearer(token): + return {"Authorization": f"Bearer {token}"} + + +def _b64(data: bytes) -> str: + return base64.urlsafe_b64encode(data).rstrip(b"=").decode() + + +def _hs256_with_key(claims, key: bytes) -> str: + """Sign by hand, so tests can forge what PyJWT itself refuses to produce.""" + import hashlib + import hmac + + header = _b64(json.dumps({"alg": "HS256", "typ": "JWT"}).encode()) + signing_input = f"{header}.{_b64(json.dumps(claims).encode())}" + sig = hmac.new(key, signing_input.encode(), hashlib.sha256).digest() + return f"{signing_input}.{_b64(sig)}" + + +def _token(claims=None, *, key=SECRET, algorithm="HS256"): + return jwt.encode(claims or {"sub": "alice"}, key, algorithm=algorithm) + + +# ------------------------------------------------------------------- JWTPlugin + + +def test_jwt_valid_token_is_decoded_onto_request(): + status, body = call_wsgi( + _app(JWTPlugin(SECRET)), "/whoami", {}, headers=_bearer(_token()) + ) + assert status.startswith("200") and body == {"sub": "alice"} + + +@pytest.mark.parametrize( + "headers", + [ + {}, + _bearer("garbage"), + _bearer(_token(key="wrong-secret-0123456789abcdef0123")), + ], + ids=["no-header", "garbage", "wrong-signature"], +) +def test_jwt_bad_tokens_get_401(headers): + status, _ = call_wsgi(_app(JWTPlugin(SECRET)), "/whoami", {}, headers=headers) + assert status.startswith("401") + + +def test_jwt_mapper_renames_claims(): + plugin = JWTPlugin(SECRET, mapper={"sub": "user_id"}) + _, body = call_wsgi(_app(plugin), "/whoami", {}, headers=_bearer(_token())) + assert body == {"user_id": "alice"} + + +def test_jwt_ignore_methods_and_ping_need_no_token(): + app = _app(JWTPlugin(SECRET, ignore_methods=["other"])) + assert call_wsgi(app, "/other", {})[0].startswith("200") + assert call_wsgi(app, "/ping", method="GET")[0].startswith("200") + assert call_wsgi(app, "/whoami", {})[0].startswith("401") + + +def test_jwt_verify_false_lets_anything_through_with_a_warning(): + app = _app(JWTPlugin(SECRET, verify=False)) + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always") + assert call_wsgi(app, "/whoami", {}, headers=_bearer("garbage"))[0].startswith( + "200" + ) + assert caught + unsigned = _token(key="another-secret-0123456789abcdef01") + _, body = call_wsgi(app, "/whoami", {}, headers=_bearer(unsigned)) + assert body == {"sub": "alice"} + + +def test_jwt_rs256_with_public_key(): + from cryptography.hazmat.primitives import serialization + from cryptography.hazmat.primitives.asymmetric import rsa + + private = rsa.generate_private_key(public_exponent=65537, key_size=2048) + public_pem = private.public_key().public_bytes( + serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo + ) + app = _app(JWTPlugin(public_pem, algorithms=["RS256"])) + genuine = _token(key=private, algorithm="RS256") + assert call_wsgi(app, "/whoami", {}, headers=_bearer(genuine))[0].startswith("200") + + +# ------------------------------------------------------------- ApiKeyAuthPlugin + + +def test_api_key_plugin(): + app = _app(ApiKeyAuthPlugin(API_KEY)) + ok = call_wsgi(app, "/other", {}, headers={"Authorization": API_KEY}) + assert ok == ("200 OK", "other") + for headers in ({}, {"Authorization": "nope"}): + assert call_wsgi(app, "/other", {}, headers=headers)[0].startswith("401") + + +# ---------------------------------------------------- mk_superadmin_middleware + + +async def _hello(request): + return web.json_response("hello") + + +def _aiohttp_get(middleware, headers): + app = web.Application(middlewares=[middleware]) + app.router.add_get("/hello", _hello) + + async def go(): + async with TestClient(TestServer(app)) as client: + resp = await client.get("/hello", headers=headers) + return resp.status + + return asyncio.run(go()) + + +def test_superadmin_middleware(): + mw = mk_superadmin_middleware(API_KEY) + assert _aiohttp_get(mw, {"Authorization": API_KEY}) == 200 + assert _aiohttp_get(mw, {"Authorization": "nope"}) == 401 + assert _aiohttp_get(mw, {}) == 401 + + +# ------------------------------------- hardening (the sprint fixed these in the +# aiohttp middleware only; JWTPlugin is what guards an mk_app service) + + +def _rsa_keys(): + from cryptography.hazmat.primitives import serialization + from cryptography.hazmat.primitives.asymmetric import rsa + + private = rsa.generate_private_key(public_exponent=65537, key_size=2048) + public_pem = private.public_key().public_bytes( + serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo + ) + return private, public_pem + + +@pytest.mark.parametrize( + "token", + [ + _token({"sub": "alice", "exp": int(time.time()) - 60}), # expired + _token(algorithm="HS512"), # an algorithm not allowed + _b64(b'{"alg":"none","typ":"JWT"}') + "." + _b64(b'{"sub":"alice"}') + ".", + ], + ids=["expired", "disallowed-alg", "alg-none"], +) +def test_jwt_rejected_tokens_are_401_not_500_and_not_echoed(token): + status, body = call_wsgi( + _app(JWTPlugin(SECRET)), "/whoami", {}, headers=_bearer(token) + ) + assert status.startswith("401") + assert token not in json.dumps(body) + + +def test_jwt_token_of_another_key_family_is_401(): + private, public_pem = _rsa_keys() + app = _app(JWTPlugin(public_pem, algorithms=["RS256"])) + assert call_wsgi(app, "/whoami", {}, headers=_bearer(_token()))[0].startswith("401") + + +def test_jwt_needs_the_bearer_scheme(): + app = _app(JWTPlugin(SECRET)) + headers = {"Authorization": f"Basic {_token()}"} + assert call_wsgi(app, "/whoami", {}, headers=headers)[0].startswith("401") + headers = {"Authorization": f"bearer {_token()}"} + assert call_wsgi(app, "/whoami", {}, headers=headers)[0].startswith("200") + + +@pytest.mark.parametrize( + "kwargs", + [ + {"secret": ""}, # would fail on every request instead + {"secret": SECRET, "algorithms": ["HS256", "RS256"]}, + {"secret": SECRET, "algorithms": ["none"]}, + ], +) +def test_jwt_unsafe_setups_are_refused_at_construction(kwargs): + with pytest.raises((ValueError, TypeError)): + JWTPlugin(**kwargs) + + +def test_jwt_verify_false_needs_no_secret(): + JWTPlugin(verify=False) + + +def test_jwt_ignore_methods_with_published_openapi(): + """Routes py2http adds itself (``/openapi``) carry no ``method_name``.""" + app = mk_app( + [whoami, other], + plugins=[JWTPlugin(SECRET, ignore_methods=["other"])], + publish_openapi=True, + ) + status, _ = call_wsgi(app, "/openapi", method="GET", headers=_bearer(_token())) + assert status.startswith("200") + assert call_wsgi(app, "/openapi", method="GET")[0].startswith("401") + + +def test_jwt_and_api_key_plugins_stack(): + app = _app(JWTPlugin(SECRET, ignore_methods=["other"]), ApiKeyAuthPlugin(API_KEY)) + headers = {"Authorization": API_KEY} + assert call_wsgi(app, "/other", {}, headers=headers)[0].startswith("200") + + +def test_cors_preflight_passes_without_token(): + app = mk_app([whoami], plugins=[JWTPlugin(SECRET)], enable_cors=True) + status, _ = call_wsgi(app, "/whoami", method="OPTIONS") + assert status.startswith("200") + + +@pytest.mark.parametrize("key", ["", None, b""]) +def test_empty_api_keys_are_refused(key): + with pytest.raises(ValueError): + ApiKeyAuthPlugin(key) + with pytest.raises(ValueError): + mk_superadmin_middleware(key) + + +# ------------------------------------- adversarial-review findings, and #25 + + +def _raw_get(app, path, raw_header: bytes): + """GET with a raw (possibly non-UTF-8) ``Authorization`` header, as WSGI hands + it over: latin-1 decoded bytes.""" + import io + from wsgiref.util import setup_testing_defaults + + environ = { + "REQUEST_METHOD": "GET", + "PATH_INFO": path, + "HTTP_AUTHORIZATION": raw_header.decode("latin-1"), + "wsgi.input": io.BytesIO(b""), + } + setup_testing_defaults(environ) + statuses = [] + b"".join(app(environ, lambda status, headers, *a: statuses.append(status))) + return statuses[0] + + +@pytest.mark.parametrize( + "plugin", + [JWTPlugin(SECRET), ApiKeyAuthPlugin(API_KEY)], + ids=["jwt", "api-key"], +) +def test_non_utf8_authorization_header_is_401_not_500(plugin): + app = mk_app([whoami], plugins=[plugin], http_method="get") + assert _raw_get(app, "/whoami", b"Bearer \xe9.\xe9.\xe9").startswith("401") + + +def _aiohttp_raw_status(middleware, raw_header: bytes) -> int: + """Send a raw (non-UTF-8) header over a socket, which aiohttp's client won't.""" + app = web.Application(middlewares=[middleware]) + app.router.add_get("/hello", _hello) + + async def go(): + async with TestServer(app) as server: + reader, writer = await asyncio.open_connection(server.host, server.port) + writer.write( + b"GET /hello HTTP/1.1\r\nHost: x\r\nConnection: close\r\n" + b"Authorization: " + raw_header + b"\r\n\r\n" + ) + await writer.drain() + status_line = await reader.readline() + writer.close() + return int(status_line.split()[1]) + + return asyncio.run(go()) + + +@pytest.mark.parametrize( + "middleware", + [mk_superadmin_middleware(API_KEY), mk_jwt_middleware(SECRET)], + ids=["superadmin", "jwt"], +) +def test_aiohttp_non_utf8_authorization_header_is_401(middleware): + assert _aiohttp_raw_status(middleware, b"Bearer \xe9.\xe9.\xe9") == 401 + + +def test_mixed_families_are_fine_without_verification(): + JWTPlugin(verify=False, algorithms=["HS256", "RS256"]) + + +def test_server_side_key_problem_is_logged(caplog): + _, public_pem = _rsa_keys() + plugin = JWTPlugin( + public_pem, algorithms=["HS256", "RS256"], allow_mixed_algorithm_families=True + ) + forged = _hs256_with_key({"sub": "mallory"}, public_pem) + with caplog.at_level("WARNING", logger="py2http"): + status, _ = call_wsgi(_app(plugin), "/whoami", {}, headers=_bearer(forged)) + assert status.startswith("401") + assert "server-side reason" in caplog.text + + +@pytest.mark.parametrize("make", [JWTPlugin, mk_jwt_middleware]) +def test_audience_passes_through_decode_kwargs(make): + """A token with an ``aud`` claim needs ``decode_kwargs={'audience': ...}`` (#25).""" + token = _token({"sub": "alice", "aud": "my-api"}) + if make is JWTPlugin: + app = _app(JWTPlugin(SECRET, decode_kwargs={"audience": "my-api"})) + assert ( + call_wsgi(app, "/whoami", {}, headers=_bearer(token))[1]["sub"] == "alice" + ) + app = _app(JWTPlugin(SECRET)) + assert call_wsgi(app, "/whoami", {}, headers=_bearer(token))[0].startswith( + "401" + ) + with pytest.raises(ValueError, match="options"): + make(SECRET, decode_kwargs={"options": {"verify_signature": False}}) diff --git a/py2http/tests/test_error_statuses.py b/py2http/tests/test_error_statuses.py new file mode 100644 index 0000000..ad43571 --- /dev/null +++ b/py2http/tests/test_error_statuses.py @@ -0,0 +1,65 @@ +"""Errors a dispatched function raises become the HTTP status that names them.""" + +import pytest +from aiohttp import web +from i2.errors import ( + AuthorizationError, + DuplicateRecordError, + ForbiddenError, + InputError, + NotFoundError, + OverwritesNotAllowed, +) + +from py2http import mk_app +from py2http.default_configs import aiohttp_error_handler, error_status +from py2http.tests.utils_for_testing import call_wsgi + +CASES = [ + (InputError("bad input"), 400), + (DuplicateRecordError("dup"), 400), + (AuthorizationError("who are you"), 401), + (ForbiddenError("not you"), 403), + (OverwritesNotAllowed("no overwrite"), 403), + (NotFoundError("nothing here"), 404), + (KeyError("internal detail"), 500), +] + + +def _app_raising(error): + def fail(): + raise error + + return mk_app([fail]) + + +@pytest.mark.parametrize("error, status", CASES) +def test_error_status(error, status): + assert error_status(error) == status + + +@pytest.mark.parametrize("error, status", CASES) +def test_bottle_app_statuses(error, status, capsys): + got, body = call_wsgi(_app_raising(error), "/fail", {}) + assert got.startswith(str(status)) + assert set(body) == {"error"} + + +def test_bottle_400_reason_names_the_error(): + got, _ = call_wsgi(_app_raising(InputError("x")), "/fail", {}) + assert got == "400 InputError" + + +def test_opaque_errors_hides_500_text(monkeypatch, capsys): + monkeypatch.setenv("OPAQUE_ERRORS", "1") + _, body = call_wsgi(_app_raising(KeyError("internal detail")), "/fail", {}) + assert body == {"error": "Internal server error"} + + +@pytest.mark.parametrize("error, status", CASES) +def test_aiohttp_error_handler(error, status): + with pytest.raises(web.HTTPException) as caught: + aiohttp_error_handler(error) + assert caught.value.status == status + if status == 500: + assert "internal detail" not in caught.value.text diff --git a/py2http/tests/test_instance_endpoints.py b/py2http/tests/test_instance_endpoints.py index bbe2072..e969618 100644 --- a/py2http/tests/test_instance_endpoints.py +++ b/py2http/tests/test_instance_endpoints.py @@ -1,19 +1,15 @@ """Instance endpoints (one object every client shares) need an owner hook and are bounded in size (#30).""" -import io -import json -from wsgiref.util import setup_testing_defaults - import pytest from i2.errors import AuthorizationError, InputError from py2http import mk_app from py2http.service import ( _get_func_to_dispatch_handler, - current_request, shared_objects, ) +from py2http.tests.utils_for_testing import as_owner, call_wsgi class Store(dict): @@ -43,12 +39,7 @@ def _instance(endpoint=None, **extras): ) -def _as(owner, func, **kwargs): - token = current_request.set(owner) - try: - return func(**kwargs) - finally: - current_request.reset(token) +_as = as_owner def test_instance_endpoint_without_owner_is_refused_at_build(): @@ -130,19 +121,7 @@ def test_custom_sizeof(): def _post(app, path, payload, user): - body = json.dumps(payload).encode() - environ = { - "REQUEST_METHOD": "POST", - "PATH_INFO": path, - "CONTENT_TYPE": "application/json", - "CONTENT_LENGTH": str(len(body)), - "HTTP_X_USER": user, - "wsgi.input": io.BytesIO(body), - } - setup_testing_defaults(environ) - statuses = [] - out = b"".join(app(environ, lambda status, headers, *a: statuses.append(status))) - return statuses[0], json.loads(out) + return call_wsgi(app, path, payload, headers={"X-User": user}) def test_http_statuses(): diff --git a/py2http/tests/test_obj_ownership.py b/py2http/tests/test_obj_ownership.py index bcaca4b..8ed233b 100644 --- a/py2http/tests/test_obj_ownership.py +++ b/py2http/tests/test_obj_ownership.py @@ -1,9 +1,5 @@ """Class-endpoint objects belong to their creator and are bounded in size (#22).""" -import io -import json -from wsgiref.util import setup_testing_defaults - import pytest from i2.errors import AuthorizationError, InputError @@ -12,9 +8,9 @@ OwnedObjStore, _get_func_to_dispatch_handler, bounded_deep_sizeof, - current_request, shared_objects, ) +from py2http.tests.utils_for_testing import as_owner, call_wsgi class Bag: @@ -33,12 +29,7 @@ def _owned(**extras): ) -def _as(owner, func, **kwargs): - token = current_request.set(owner) - try: - return func(**kwargs) - finally: - current_request.reset(token) +_as = as_owner def test_class_endpoint_without_owner_is_refused_at_build(): @@ -129,20 +120,7 @@ def test_owned_store_rejects_bad_bounds(): def _post(app, path, payload, user): - body = json.dumps(payload).encode() - environ = {} - setup_testing_defaults(environ) - environ.update( - REQUEST_METHOD="POST", - PATH_INFO=path, - CONTENT_TYPE="application/json", - CONTENT_LENGTH=str(len(body)), - HTTP_X_USER=user, - ) - environ["wsgi.input"] = io.BytesIO(body) - statuses = [] - out = b"".join(app(environ, lambda status, headers, *a: statuses.append(status))) - return statuses[0], json.loads(out) + return call_wsgi(app, path, payload, headers={"X-User": user}) def test_owner_hook_reads_the_real_request(): @@ -251,3 +229,28 @@ def churn(i): with ThreadPoolExecutor(8) as pool: list(pool.map(churn, range(8))) assert store.total_bytes == sum(store.owner_bytes(o) for o in range(9)) + + +@pytest.mark.parametrize( + "kwargs, match", + [ + ({"_attr_name": "add"}, "_attr_name must be None"), + ({"_obj_id": ""}, "_attr_name must be provided"), + ], +) +def test_misdirected_calls_are_input_errors(kwargs, match): + func = _owned() + if kwargs.get("_obj_id") == "": + kwargs["_obj_id"] = _as("ann", func) + with pytest.raises(InputError, match=match): + _as("ann", func, **kwargs) + + +def test_shared_mode_custom_store_drops_an_oversized_object(): + store = {} + func = _owned(obj_owner=shared_objects, obj_store=store, max_obj_size=5_000) + bag = _as("ann", func) + assert list(store) == [bag] + with pytest.raises(InputError, match="size limit"): + _as("bob", func, _obj_id=bag, _attr_name="add", item="y" * 10_000) + assert store == {} diff --git a/py2http/tests/test_positional_only.py b/py2http/tests/test_positional_only.py index 02e563e..819ac94 100644 --- a/py2http/tests/test_positional_only.py +++ b/py2http/tests/test_positional_only.py @@ -1,14 +1,11 @@ """Positional-only parameters (``dict.__getitem__``, ``def f(x, /)``) can be called over JSON, which supplies arguments by name (i2mint/py2http#13, #14).""" -import io -import json -from wsgiref.util import setup_testing_defaults - import pytest from i2.errors import InputError from py2http import mk_app +from py2http.tests.utils_for_testing import call_wsgi from py2http.util import call_with_inputs @@ -64,19 +61,7 @@ def no_signature(func): assert d == {"b": 2} -def _post(app, path, payload): - body = json.dumps(payload).encode() - environ = { - "REQUEST_METHOD": "POST", - "PATH_INFO": path, - "CONTENT_TYPE": "application/json", - "CONTENT_LENGTH": str(len(body)), - "wsgi.input": io.BytesIO(body), - } - setup_testing_defaults(environ) - statuses = [] - out = b"".join(app(environ, lambda status, headers, *a: statuses.append(status))) - return statuses[0], json.loads(out) +_post = call_wsgi def test_dict_dunders_over_http(): diff --git a/py2http/tests/utils_for_testing.py b/py2http/tests/utils_for_testing.py index 87134e0..d997741 100644 --- a/py2http/tests/utils_for_testing.py +++ b/py2http/tests/utils_for_testing.py @@ -1,6 +1,13 @@ +"""Helpers shared by py2http's tests: calling a WSGI app in-process, and running a +dispatched function as if a given request (or owner) were being handled.""" + +import io +import json from contextlib import contextmanager from multiprocessing.context import Process from time import sleep +from wsgiref.util import setup_testing_defaults + from py2http.util import conditional_logger, deprecate @@ -8,7 +15,6 @@ @deprecate def run_server(launcher, wait_before_entering=0, verbose=False, **kwargs): """Context manager to launch server on entry, and shut it down on exit""" - from warnings import warn clog = conditional_logger(verbose) server = None @@ -24,3 +30,40 @@ def run_server(launcher, wait_before_entering=0, verbose=False, **kwargs): clog(f"Terminating server...") server.terminate() clog(f"... server terminated") + + +def call_wsgi(app, path, payload=None, *, method="POST", headers=()): + """Send one request to the WSGI ``app`` in-process; return ``(status, body)``, the + body decoded from JSON when it is JSON. + + ``headers`` maps header names (``"Authorization"``) to values. + """ + body = b"" if payload is None else json.dumps(payload).encode() + environ = { + "REQUEST_METHOD": method, + "PATH_INFO": path, + "CONTENT_TYPE": "application/json", + "CONTENT_LENGTH": str(len(body)), + "wsgi.input": io.BytesIO(body), + } + for name, value in dict(headers).items(): + environ["HTTP_" + name.upper().replace("-", "_")] = value + setup_testing_defaults(environ) + statuses = [] + out = b"".join(app(environ, lambda status, headers, *a: statuses.append(status))) + try: + return statuses[0], json.loads(out) + except ValueError: + return statuses[0], out.decode() + + +def as_owner(owner, func, **kwargs): + """Call ``func(**kwargs)`` while ``owner`` is the request being handled, so an + ``obj_owner=lambda req: req`` hook sees ``owner``.""" + from py2http.service import current_request + + token = current_request.set(owner) + try: + return func(**kwargs) + finally: + current_request.reset(token)