"""Generate docs/openapi.json from docs/api-inventory.json, and refresh that inventory from the routes the app actually serves. The inventory is a list of {surface, endpoints:[{method,path,auth,params,purpose, example}]} objects (one per API surface). Two halves, maintained differently: STRUCTURE (surface, method, path, auth) is DERIVED. It is walked out of the live url_map and the decorators on each view, so it cannot drift from the code. It had: this branch carried no inventory at all, so the generator had no input, the parity test skipped, and 24 routes served traffic documented nowhere. PROSE (purpose, params, example) is HUMAN. What an endpoint is FOR cannot be derived, so a refresh never overwrites text that is already there - it fills in a new route from its docstring and leaves every existing entry's words alone. So: add a route, run --refresh, and replace the seeded prose with the reason the route exists. Coverage is enforced on top of that by tests/test_api_inventory_parity.py, which fails when a served route has no entry AND when a committed auth string disagrees with what this file re-derives. Served interactively at /api/docs (see shopdb/core/api/docs.py). venv/bin/python scripts/gen_openapi.py # spec from the inventory venv/bin/python scripts/gen_openapi.py --refresh # inventory first, then spec THE AUTH STRING IS A GRAMMAR, not free text. Its first word names the mechanism and everything after it is prose; '-optional' on that word means the gate can be satisfied without a credential. See AUTH_SCHEME and _security. Derived strings come from DECORATOR_AUTH, HANDLER_AUTH and INLINE_AUTH below, so a route whose gate this cannot see says so rather than guessing. KNOWN LIMITATION: QUERY PARAMETERS ARE NOT DERIVED. Path arguments come out of the rule, and a body is declared for the verbs that take one, but nothing here reads request.args out of a handler - so every route whose real interface is a query string ('?q=', '?page=', '?status=') publishes params as 'not documented' and an operation with no `parameters` beyond its path. That is a gap in the generated spec, not in the route: a machine client built from this document can call those endpoints but cannot filter, search or page them. Filling it means either a handler sweep for request.args reads - which cannot see a key built at runtime, so it would be right most of the time and silently wrong the rest - or writing the query contract into the inventory's PROSE half by hand, where a human owns it. The prose half is the honest option and is where it should land. The one query credential that IS expressed is the kiosk setup code, because it is auth rather than a filter and rides in as a securityScheme. """ import json import os import re import sys HERE = os.path.dirname(os.path.abspath(__file__)) REPO = os.path.dirname(HERE) INVENTORY = os.path.join(REPO, 'docs', 'api-inventory.json') OUT = os.path.join(REPO, 'docs', 'openapi.json') VERBS = ('get', 'post', 'put', 'patch', 'delete', 'head', 'options') def _servers(): """Servers block for the spec: the relative mount, plus this site's own URL if one was supplied. SHOPDB_PUBLIC_URL is read from the environment rather than stored, because the generated spec is published to a public wiki - a site URL baked into the generator ends up in everyone's documentation, including sites it is wrong for. """ servers = [] siteurl = (os.environ.get('SHOPDB_PUBLIC_URL') or '').strip().rstrip('/') if siteurl: servers.append({'url': siteurl, 'description': 'this site'}) servers.append({'url': '/', 'description': 'relative to the deployed mount'}) return servers def _product_version(): """Read __version__ out of shopdb/__init__.py without importing the app. Importing shopdb here would pull in Flask, the plugin loader and a database configuration, none of which this generator needs. """ here = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) init = os.path.join(here, 'shopdb', '__init__.py') with open(init, 'r', encoding='utf-8') as handle: match = re.search(r"^__version__ = '([^']+)'", handle.read(), re.M) if not match: raise SystemExit('could not read __version__ from ' + init) return match.group(1) def _norm_path(p): # Flask / -> OpenAPI {id} normalised = re.sub(r'<(?:[^:>]+:)?([^>]+)>', r'{\1}', p or '') # A trailing slash in a published path is a distinct resource to a strict # proxy (IIS), which answers the unslashed URL with a redirect a client # behind a subpath mount cannot follow. Emit the canonical, slashless form. if len(normalised) > 1: normalised = normalised.rstrip('/') return normalised # The machine-readable half of an auth string is its FIRST word. Everything # after it is prose for a human reader. An optional gate is spelled by suffixing # that word with '-optional'. # # Substring matching is what this replaces, and it is why nine operations # published `security: []` - "no credentials needed" - while their handler # required one. 'none (single-use bootstrap token...)' starts with 'none', so # the old sweep published the two bootstrap routes as open, and one of them # hands out the installable agent package and spends a use. A leading token # cannot be fooled by the prose that explains it. AUTH_SCHEME = { 'jwt': 'bearerAuth', 'permission': 'bearerAuth', 'api-key': 'apiKeyAuth', 'setup-code': 'setupCodeAuth', } OPTIONAL_SUFFIX = '-optional' def _auth_head(auth): """The mechanism word at the front of an auth string, lowercased.""" match = re.match(r'[a-z-]+', (auth or '').strip().lower()) return match.group(0) if match else '' def _security(auth): """OpenAPI security for one endpoint's documented auth string. Optional auth is the interesting case. It means "works without a credential, does more with one", which OpenAPI expresses as a list carrying BOTH the empty requirement and the scheme. Two shapes reach it: jwt-optional works signed out; the handler decides what shows. api-key-optional token OR network. A kiosk admits either the per-device token or a source address inside the site's CIDR allowlist, so an allowlisted caller sends no credential. Publishing that as `[]` is not a lie, but it is not the truth either: it tells a reader there is no token to send, and the allowlist is invisible. Optional says both halves exist. An unrecognised word publishes as bearer-required, which is the safe direction to be wrong in: a reader who brings a token they did not need loses nothing, a reader told none was needed is locked out on the day. """ head = _auth_head(auth) optional = head.endswith(OPTIONAL_SUFFIX) if optional: head = head[:-len(OPTIONAL_SUFFIX)] if not head or head in ('none', 'public'): return [] scheme = AUTH_SCHEME.get(head, 'bearerAuth') if optional: return [{scheme: []}, {}] return [{scheme: []}] def _path_parameters(rawpath): """Path parameters, typed from the Flask converter that declared them. The generator emitted no `parameters` at all, which makes the document invalid OpenAPI 3.1 - and breaks its only machine consumer outright, because a tool built from an operation with no parameters has nowhere to put the id the caller supplied, so it is dropped in silence. """ parameters = [] for converter, name in re.findall(r'<(?:([^:>]+):)?([^>]+)>', rawpath or ''): schema = {'type': 'integer'} if converter in ('int',) else {'type': 'string'} if converter == 'float': schema = {'type': 'number'} if converter == 'path': schema = {'type': 'string', 'description': 'may contain slashes'} parameters.append({ 'name': name, 'in': 'path', 'required': True, 'schema': schema, }) return parameters def _request_body(verb, entry): """A JSON body for the verbs that take one. The inventory describes bodies in prose, not as schemas, so this does not invent field names it cannot verify - it declares an object and carries the prose. That is honest, and it is the difference between a machine client that can send a body and one that cannot send anything at all. """ if verb not in ('post', 'put', 'patch'): return None described = entry.get('params') or '' return { 'required': verb in ('post', 'put'), 'content': { 'application/json': { 'schema': { 'type': 'object', 'additionalProperties': True, 'description': described or 'See the endpoint description.', } } }, } def _responses(security, parameters): """The envelope, plus the failures a caller has to handle. One hardcoded 200 was the whole response section, so a generated client had no idea any call could fail, and 19 operations that do not return the envelope at all were documented as if they did. """ responses = { '200': { 'description': 'Success. Body is the success_response envelope: ' '{status, data, meta}.', 'content': {'application/json': { '$ref': '#/components/schemas/SuccessEnvelope'}}, }, 'default': { 'description': 'Error. Body is the error envelope; the code and ' 'message are nested under data.error.', 'content': {'application/json': { '$ref': '#/components/schemas/ErrorEnvelope'}}, }, } # An endpoint that can be called without a token cannot answer 401. if security and security != [{}]: requires = not any(item == {} for item in security) if requires: responses['401'] = {'description': 'Missing or invalid credentials.'} responses['403'] = {'description': 'Authenticated, but not permitted.'} if parameters: responses['404'] = {'description': 'No such record.'} return responses def _summary(text, fallback): """A summary that ends on a word. Cutting at a fixed 120 characters truncated 95 of them mid-word, which is what a tool picker shows a user as the whole description of the call. """ source = (text or fallback or '').strip() if len(source) <= 120: return source head = source[:120] cut = head.rfind(' ') # Trailing connectives read worse than a clean cut: "servicelevel +..." # promises a continuation the reader will never see. return (head[:cut] if cut > 40 else head).rstrip(' ,;:.+-/&|(') + '...' def build(surfaces): paths = {} tags = [] for s in surfaces: tags.append({'name': s['surface']}) for e in s.get('endpoints', []): # 'PUT|PATCH' is how the inventory writes a multi-verb route. # Without the pipe here, neither verb matched VERBS and the # operation vanished from the spec without a word. for verb in re.split(r'[\/,|]', (e.get('method') or 'GET')): verb = verb.strip().lower() if verb not in VERBS: continue path = _norm_path(e.get('path')) if not path: continue desc = [] if e.get('purpose'): desc.append(e['purpose']) if e.get('auth'): desc.append('\n\n**Auth:** ' + e['auth']) if e.get('params'): desc.append('\n\n**Params:** ' + e['params']) if e.get('example'): desc.append('\n\n**Example:**\n```\n' + e['example'] + '\n```') security = _security(e.get('auth')) parameters = _path_parameters(e.get('path')) operation = { 'tags': [s['surface']], 'summary': _summary(e.get('purpose'), path), 'description': ''.join(desc), 'security': security, 'responses': _responses(security, parameters), } if parameters: operation['parameters'] = parameters body = _request_body(verb, e) if body: operation['requestBody'] = body paths.setdefault(path, {})[verb] = operation return { 'openapi': '3.1.0', 'info': { 'title': 'ShopDB Flask API', # Read from the code, not restated. A hardcoded copy here had # already drifted a release behind, and the same mistake in the # installer script shipped an exe stamped with the wrong version. 'version': _product_version(), 'description': ( 'Asset-management API (core + plugins). Responses use a ' '`success_response` envelope: `{status, data, meta}`. Auth: Bearer ' 'JWT (login or a managed PAT) for `jwt`/`admin`/`permission:*`; ' '`X-API-Key` for collector/managed-token endpoints; public endpoints ' 'need neither. A permission-gated route is published as ' 'bearer-required, which is what a default install does - a site ' 'that grants the permission to the `guest` role makes that same ' 'route answer a caller with no token.'), }, # One site's production hostname was hardcoded here, so every generated # spec published it to the public wiki and offered a second site a # server it cannot reach. The relative mount is the only server this # generator can honestly name; a site that wants its own in the spec # sets SHOPDB_PUBLIC_URL when regenerating. 'servers': _servers(), 'components': { 'schemas': { # The envelope every JSON endpoint returns. Worth spelling out # because the error shape nests one level deeper than most # people assume, and code written against the assumption reads # undefined on every error it tries to report. 'SuccessEnvelope': { 'type': 'object', 'properties': { 'status': {'type': 'string', 'enum': ['success']}, 'data': {'description': 'The payload. Shape is per endpoint.'}, 'meta': { 'type': 'object', 'properties': { 'timestamp': {'type': 'string', 'format': 'date-time'}, 'requestid': {'type': 'string'}, 'pagination': { 'type': 'object', 'properties': { 'page': {'type': 'integer'}, 'perpage': {'type': 'integer'}, 'total': {'type': 'integer'}, 'pages': {'type': 'integer'}, }, }, }, }, }, 'required': ['status'], }, 'ErrorEnvelope': { 'type': 'object', 'properties': { 'status': {'type': 'string', 'enum': ['error']}, 'data': { 'type': 'object', 'properties': { 'error': { 'type': 'object', 'properties': { 'code': {'type': 'string'}, 'message': {'type': 'string'}, 'details': {'type': 'object'}, }, 'required': ['code', 'message'], }, }, 'required': ['error'], }, 'meta': {'type': 'object'}, }, 'required': ['status', 'data'], }, }, 'securitySchemes': { 'bearerAuth': {'type': 'http', 'scheme': 'bearer', 'bearerFormat': 'JWT', 'description': 'Login token or a managed Personal Access Token (scoped).'}, # One scheme covers every non-JWT header credential in this API, # because the header NAME varies per surface and each operation's # **Auth:** line says which one it wants. Publishing a scheme per # header would be tidier and no more informative. 'apiKeyAuth': {'type': 'apiKey', 'in': 'header', 'name': 'X-API-Key', 'description': 'A managed service or device token in a header. ' 'X-API-Key for collector and managed tokens ' '(collector.ingest, geenforce.fetch); ' 'X-Endpoint-Bootstrap-Token for agent bootstrap; ' 'X-Kiosk-Token for a kiosk screen. The operation ' 'description names the header and, where the gate ' 'is token-OR-network, the allowlist that is the ' 'other half.'}, 'setupCodeAuth': {'type': 'apiKey', 'in': 'query', 'name': 'code', 'description': 'A single-use provisioning code, redeemed in the ' 'query string because the caller is a browser being ' 'pointed at a URL. Spent on any attempt, right or ' 'wrong, and short-lived.'}, }}, 'tags': tags, 'paths': paths, } # --------------------------------------------------------------------------- # The derived half: refresh the inventory from the live url_map (--refresh) # --------------------------------------------------------------------------- # Only the API. The frontend shell and its assets are served by this app but are # not the contract this document describes, and tests/test_api_inventory_parity # walks the same /api-only surface - an entry for anything else reports there as # a documented route that is not served. API_PREFIX = '/api' # The auth string for a route whose gate this cannot see. Deliberately not # 'none': an unclassified route is the one case where a guess writes a lie into # the published contract, and a site reads that contract to decide what its # guest role and its firewall may reach. This greps, and it renders at /api/docs. # Hand-editing the inventory does NOT clear this: auth is derived, so the next # --refresh writes it straight back. Teaching the sweep below is the fix, which # is also the only way the classification survives the route moving. NEEDS_REVIEW = ('review: no gate found on the view - teach the sweep in ' 'scripts/gen_openapi.py, or gate the view') # The jwt-optional wording, in one place because three branches reach it. JWT_OPTIONAL = 'jwt-optional (works signed out; the handler decides what shows)' # Decorators that ARE the gate. require_permission and require_role are not # here: they stamp _requires_permission onto the wrapper, which is read first # and carries the permission name itself. DECORATOR_AUTH = { 'require_api_key': 'api-key (per-plugin collector token)', 'require_api_key_legacy': 'api-key (collector token, legacy shared key)', 'require_fetch_token': 'api-key (geenforce fetch token, or the IP allowlist)', 'require_report_token': 'api-key (geenforce report token)', } # Gates that live INSIDE the handler, found by the call the handler makes. An # agent, a kiosk screen and a bootstrapping PC each authenticate on something # that is not a JWT, so none of them can be gated by a decorator - and from # outside the view, an inline gate and no gate at all look exactly alike. That # is how 165 GET routes came to answer anybody. HANDLER_AUTH = ( ('_agent_auth_or_error', 'api-key (endpoint agent credential, or the fleet ' 'key while sharedkeyallowed is on)'), ('authenticate_agent', 'api-key (endpoint agent credential, or the fleet ' 'key while sharedkeyallowed is on)'), ('collector_ingest', 'api-key (endpoints collector token)'), ('_check_collector_auth', 'api-key (per-plugin collector token)'), ('collector_authorized', 'api-key (per-plugin collector token)'), # Token OR network, either sufficient, so neither 'none' nor 'api-key' is # the truth on its own: an allowlisted PC sends no credential at all, and a # screen by the cabinet sends the device token from anywhere. 'none' was # published here, which told a reader there is no token to send and left # the allowlist invisible. Optional says both halves exist. ('_kiosk_open', 'api-key-optional (kiosk: the per-device token in the ' 'X-Kiosk-Token header, OR a caller address inside the ' "site's kiosk CIDR allowlist - an allowlisted caller sends " "no credential; 404 while the site's kiosk flag is off)"), # REQUIRED, not 'none'. The handler reads the token and 404s without it, # and /bootstrap/package hands out the installable agent package and spends # a use doing it. ('_presented_token', 'api-key (single-use bootstrap token in the ' 'X-Endpoint-Bootstrap-Token header, or ?t= while that ' 'is enabled; per-IP rate limited, 404 when not usable)'), # The code IS the credential: unguessable, spent on any attempt, and gone in # minutes. No header and no session, because the caller is a browser being # pointed at a URL by an administrator. ('_redeem_kiosk_setup_code', 'setup-code (single-use kiosk setup code in ' '?code=, issued by the permission-gated ' 'provisioning link; spent on any attempt, 410 ' 'once used or expired)'), ('_bootstrap_allowed', 'none (first run only: refused once an admin exists)'), ('verify_jwt_in_request', 'jwt'), ) # Routes whose gate is inline AND conditional, where the call sweep above would # name the wrong thing. One line each, and the line is the point: these are the # handlers where "what authenticates this?" has an answer no sweep can derive. INLINE_AUTH = { # The key may be absent, or invalid, and the call still proceeds: a PC # installed without the shared key must still be able to reach an admin. A # valid key only decides whether auto-enroll may issue on a clean match. 'endpoints.enroll_endpoint': 'none (per-IP rate limited and capped; a valid ' 'fleet key only decides auto-enroll)', # The unguessable token in the path IS the authorisation - the recipient of # a secure send has no account here. 'tools.peek_secret': 'none (the unguessable token in the path is the ' 'authorisation)', 'tools.redeem_secret': 'none (the unguessable token in the path is the ' 'authorisation; redeeming spends the secret)', # This IS the auth surface, so a token cannot gate it. The bounds it carries # instead are a per-IP rate limit and the 5-strike account lockout. 'auth.login': 'none (the login surface itself; rate limited, 5-strike lockout)', } # Verbs that carry a body, for the seeded params line. BODY_VERBS = ('POST', 'PUT', 'PATCH') def _literal_set(filename, name): """One module-level set literal, read without importing the module. Imported, tests/test_core/test_authz.py builds a second app at import time, which rebinds the SQLAlchemy extension out from under the one this script is walking. """ import ast try: with open(filename, 'r', encoding='utf-8') as handle: tree = ast.parse(handle.read()) except (OSError, SyntaxError): return set() for node in tree.body: if not isinstance(node, ast.Assign): continue for target in node.targets: if isinstance(target, ast.Name) and target.id == name: try: return set(ast.literal_eval(node.value)) except ValueError: return set() return set() def _reviewed_exceptions(): """(endpoints, blueprints) already reviewed as needing no permission. Read from the two sweeps in tests/ rather than restated here, so the published contract cannot disagree with the gate that enforces it. A route on one of those lists was argued about in front of somebody, in a file whose whole job is to make that argument visible; a route on neither, with no gate, is marked for review instead. That is the difference between a decision and an oversight, which is what those lists exist to keep. """ tests = os.path.join(REPO, 'tests', 'test_core') reads = os.path.join(tests, 'test_public_read_surface.py') writes = os.path.join(tests, 'test_authz.py') endpoints = (_literal_set(reads, 'PUBLIC_ENDPOINTS') | _literal_set(writes, 'EXEMPT_ENDPOINTS')) return endpoints, _literal_set(writes, 'EXEMPT_BLUEPRINTS') def _source_index(filename, _cache={}): """{function name: (decorator sources, call sources)} for one file. Read from the source, not from the live object: functools.wraps copies the name and the docstring onto the wrapper and nothing else, so by the time a view is in url_map its decorators are invisible. _requires_permission survives only because require_permission deliberately stamps it on. """ import ast if filename in _cache: return _cache[filename] index = {} try: with open(filename, 'r', encoding='utf-8') as handle: tree = ast.parse(handle.read()) except (OSError, SyntaxError): _cache[filename] = index return index for node in ast.walk(tree): if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)): continue index[node.name] = ( [ast.unparse(item) for item in node.decorator_list], [ast.unparse(item) for item in ast.walk(node) if isinstance(item, ast.Call)], ) _cache[filename] = index return index def _view_source(view): """(decorators, calls) for the undecorated view behind a url_map entry.""" import inspect original = inspect.unwrap(view) filename = inspect.getsourcefile(original) or '' return _source_index(filename).get(original.__name__, ([], [])) def _auth_for(view, endpoint, reviewed, blueprints): """The honest auth string for one route, derived in gate order. Nothing here guesses. Every branch names a mechanism that is in the code - a marker the decorator stamped on, a decorator in the source, a call the handler makes, or a name on a reviewed exemption list - and a route that matches none of them says exactly that. """ marker = getattr(view, '_requires_permission', None) if marker: roles = [name.split(':', 1)[1] for name in marker if name.startswith('role:')] if roles: return 'jwt + role:' + ' or '.join(sorted(roles)) return 'permission:' + ' or '.join(marker) decorators, calls = _view_source(view) for text in decorators: name = text.split('(', 1)[0].strip().split('.')[-1] if name in DECORATOR_AUTH: return DECORATOR_AUTH[name] if name == 'jwt_required': if 'refresh=True' in text: return 'jwt (refresh token)' return JWT_OPTIONAL if 'optional=True' in text else 'jwt' if endpoint in INLINE_AUTH: return INLINE_AUTH[endpoint] for call in calls: tail = call.split('(', 1)[0].split('.')[-1] for needle, auth in HANDLER_AUTH: if tail != needle: continue # Optional verification is a read that works signed out, not a gate. # Only the required form authenticates anybody. if needle == 'verify_jwt_in_request' and 'optional' in call: return JWT_OPTIONAL return auth blueprint, _, shortname = endpoint.partition('.') if endpoint in reviewed or shortname in reviewed or blueprint in blueprints: return 'none (reviewed: named in the public-read or mutation sweep)' return NEEDS_REVIEW def _surface_for(view, endpoint): """The tag this route is grouped under: one surface per plugin, and one per core blueprint. Derived, not preserved: a route that moves into a plugin moves tag with it, which is the whole reason the reader is looking at the tag. """ import inspect module = getattr(inspect.unwrap(view), '__module__', '') or '' if module.startswith('plugins.'): return 'plugin-' + module.split('.')[1] return 'core-' + endpoint.partition('.')[0] def _seed_purpose(view, endpoint): """First paragraph of the view's docstring, as a starting point. A docstring says what the handler does; the inventory wants what the endpoint is FOR, which is usually more. This seeds the entry so a new route is never published blank, and the prose is preserved from then on. """ import inspect doc = inspect.getdoc(inspect.unwrap(view)) or '' first = ' '.join(doc.split('\n\n')[0].split()) return first or 'UNDOCUMENTED: %s has no docstring.' % endpoint def _path_arguments(rule): """[(name, converter)] for one rule, in the order they appear.""" return [(name, converter or 'string') for converter, name in re.findall(r'<(?:([^:>]+):)?([^>]+)>', rule)] def _seed_params(rule, methods): parts = [] arguments = _path_arguments(rule) if arguments: parts.append('path: ' + ', '.join( '%s (%s)' % (name, converter) for name, converter in arguments)) if any(method in BODY_VERBS for method in methods): parts.append('body: JSON (fields not yet documented)') # Path arguments are in the rule; query arguments are read out of the # handler and cannot be derived, so an entry that has neither says it does # not know rather than claiming the route takes nothing. return '; '.join(parts) if parts else 'not documented' def _seed_example(rule, methods, auth): """A curl that is the right SHAPE for this route: its verb, a filled path and the header its auth actually wants. Placeholder ids, because the generator has no database - but a reader who substitutes one gets a call that works, which is more than a blank field gives them. """ def fill(match): converter = (match.group(1) or '').split('(')[0] if converter.startswith(('int', 'float')): return '1' return 'example.png' if converter == 'path' else 'x' url = 'http://localhost:5001' + re.sub(r'<(?:([^:>]+):)?[^>]+>', fill, rule) method = sorted(methods)[0] if methods else 'GET' command = 'curl' if method == 'GET' else 'curl -X ' + method header = '' if auth.startswith(('permission:', 'jwt')): header = ' -H "Authorization: Bearer $TOK"' elif auth.startswith('api-key'): # The header name varies per surface (X-API-Key, X-Kiosk-Token, # X-Endpoint-Bootstrap-Token), and the auth string names the one this # route wants. A pasted example with the wrong header is worse than # none: it looks like it should work. named = re.search(r'\bX-[A-Za-z0-9-]+\b', auth) header = ' -H "%s: $KEY"' % (named.group(0) if named else 'X-API-Key') elif auth.startswith('setup-code'): url += '?code=$CODE' body = '' if any(verb in methods for verb in BODY_VERBS): body = " -H 'Content-Type: application/json' -d '{}'" return '%s %s%s%s' % (command, url, header, body) def _served_routes(app): """Every /api rule, as (path, sorted methods, endpoint, view). HEAD and OPTIONS are Werkzeug's, not the product's. A rule with several verbs stays one entry, written 'PUT|PATCH', which is how the inventory has always spelled it and what the parity test splits on. """ routes = [] for rule in app.url_map.iter_rules(): if rule.endpoint == 'static' or not rule.rule.startswith(API_PREFIX): continue methods = sorted((rule.methods or set()) - {'HEAD', 'OPTIONS'}) if not methods: continue routes.append((rule.rule, methods, rule.endpoint, app.view_functions[rule.endpoint])) return sorted(routes, key=lambda row: (row[0], row[1])) def _existing_prose(): """{(path, METHOD): entry} for whatever the inventory already says. Keyed per method rather than per entry so prose survives a route that gains or loses a verb: 'PUT' still finds the words written when the entry said 'PUT|PATCH'. """ prose = {} if not os.path.isfile(INVENTORY): return prose with open(INVENTORY, 'r', encoding='utf-8') as handle: surfaces = json.load(handle) for surface in surfaces: for entry in surface.get('endpoints', []): path = entry.get('path') or '' for method in re.split(r'[/,|]', entry.get('method') or 'GET'): method = method.strip().upper() if method: prose.setdefault((path, method), entry) return prose def _inventory_app(): """The app, with every bundled plugin enabled. The inventory describes the PRODUCT, not one site's switched-on subset: a plugin disabled on the machine that happens to run this would drop its whole surface out of the published contract without a word, and the next refresh somewhere else would put it back. So a disabled plugin is refused here rather than quietly omitted. """ import sys # .env BEFORE create_app, not after: config is read while the app is built, # so a load underneath it changes nothing. A plain `python scripts/...` run # gets neither wsgi.py's load nor the flask CLI's, and falls back to the # hardcoded root@localhost URL - which surfaces as a bare 1045 access-denied # that says nothing about the real cause. from dotenv import load_dotenv load_dotenv(os.path.join(REPO, '.env')) os.environ.setdefault('FLASK_ENV', 'testing') if REPO not in sys.path: sys.path.insert(0, REPO) from shopdb import create_app app = create_app('testing') plugins = os.path.join(REPO, 'plugins') bundled = sorted( name for name in os.listdir(plugins) if os.path.isfile(os.path.join(plugins, name, 'manifest.json'))) manager = app.extensions.get('plugin_manager') off = [name for name in bundled if not (manager and manager.registry.is_enabled(name))] if off: raise SystemExit( 'these bundled plugins are disabled, so their routes would be ' 'missing\nfrom the inventory: %s\nenable them and re-run:\n' ' FLASK_APP=shopdb venv/bin/flask plugins enable %s' % (', '.join(off), off[0])) return app def refresh_inventory(): """Rewrite the inventory's derived half from the routes the app serves. Structure (surface, method, path, auth) is replaced outright. Prose (purpose, params, example) is carried over untouched wherever an entry for that route already exists, and seeded from the view's docstring where one does not. """ app = _inventory_app() reviewed, blueprints = _reviewed_exceptions() prose = _existing_prose() surfaces = {} seeded = [] unclassified = [] for path, methods, endpoint, view in _served_routes(app): auth = _auth_for(view, endpoint, reviewed, blueprints) if auth == NEEDS_REVIEW: unclassified.append('%s %s (%s)' % ('|'.join(methods), path, endpoint)) known = prose.get((path, methods[0])) if known is None: seeded.append('%s %s' % ('|'.join(methods), path)) entry = dict(known or {}) entry.update({ 'method': '|'.join(methods), 'path': path, 'auth': auth, 'params': entry.get('params') or _seed_params(path, methods), 'purpose': entry.get('purpose') or _seed_purpose(view, endpoint), 'example': entry.get('example') or _seed_example(path, methods, auth), }) # Fixed key order, so a refresh that changes nothing changes no bytes. ordered = {key: entry[key] for key in ('method', 'path', 'auth', 'params', 'purpose', 'example')} ordered.update({key: value for key, value in entry.items() if key not in ordered}) surfaces.setdefault(_surface_for(view, endpoint), []).append(ordered) document = [{'surface': name, 'endpoints': entries} for name, entries in sorted(surfaces.items())] with open(INVENTORY, 'w', encoding='utf-8') as handle: json.dump(document, handle, indent=2) handle.write('\n') total = sum(len(surface['endpoints']) for surface in document) print('refreshed %s: %d entries in %d surfaces (%d new)' % (INVENTORY, total, len(document), len(seeded))) for line in seeded: print(' seeded %s' % line) if unclassified: print('%d route(s) have no gate this can see. Classify each one by hand ' 'in the inventory, or gate it:' % len(unclassified)) for line in unclassified: print(' review %s' % line) def main(): if '--refresh' in sys.argv[1:]: refresh_inventory() with open(INVENTORY, 'r', encoding='utf-8') as handle: surfaces = json.load(handle) spec = build(surfaces) with open(OUT, 'w', encoding='utf-8') as handle: json.dump(spec, handle, indent=1) ops = sum(len(v) for v in spec['paths'].values()) print('wrote %s: %d paths, %d operations from %d surfaces' % (OUT, len(spec['paths']), ops, len(surfaces))) if __name__ == '__main__': main()