diff --git a/docs/dev/rest-api/authentication.md b/docs/dev/rest-api/authentication.md index 37310abd..9f65c18f 100644 --- a/docs/dev/rest-api/authentication.md +++ b/docs/dev/rest-api/authentication.md @@ -205,6 +205,31 @@ When the server rejects a request for authentication reasons, the response has a In your client, treat any `401` as an authentication problem and parse *both* shapes, because the standard error body is not always present. +## Rate limit rejections + +An API key that carries a rate limit gets `429 Too Many Requests` once it goes over, from the +same middleware and in the same shape as the `401` bodies above, with a `Retry-After` header in +seconds: + +```json +{ + "status": 429, + "message": "Rate limit exceeded." +} +``` + +A limit comes from `serve.tiers..rate_limit` or from the API key's own +`serve.users..rate_limit`, in requests per hour. An API key with neither has no limit, and +`GET /api/health` is never limited. See +[dashboard rate limits](dashboard.md#rate-limits) for the config and the counters, and +[services configuration](../../reference/configuration/services.md) for the keys themselves. + +!!! warning "`429` is not the `RATE_LIMITED` error body" + `RATE_LIMITED` in [Errors](errors.md) is a *service* rate limit that a handler raises, and it + arrives in the standard envelope with an `error_code`. The `429` here comes from the + middleware before any handler runs, and has the integer-`status` shape shown above. A client + that reads `error_code` to classify a `429` must cope with it being absent. + ## Access controls on top of the API key After the server authenticates a request, several additional controls decide what that specific API key can do. @@ -283,9 +308,11 @@ With `--caddy`, unshackle also starts `caddy run` using the `Caddyfile` located | Query-parameter alternative | `?secret_key=` on `GET /api/download/jobs/{job_id}/events` only | | Default bind address | `127.0.0.1:8786` | | Exempt endpoint | `GET /api/health` | -| Failure status | `401 Unauthorized` | +| Failure status | `401 Unauthorized`, or `429 Too Many Requests` for a key over its rate limit | | Missing-header body | `{"status": 401, "message": "Secret Key is Empty."}` | | Invalid-key body | `{"status": 401, "message": "Secret Key is Invalid."}` | +| Over-limit body | `{"status": 429, "message": "Rate limit exceeded."}`, plus a `Retry-After` header | +| Rate limit config | `serve.tiers..rate_limit`, overridden by `serve.users..rate_limit` | | Disable auth | `--no-key` (all requests allowed) | | Config location | `serve.api_secret` and `serve.users` in `unshackle.yaml` | diff --git a/docs/dev/rest-api/dashboard.md b/docs/dev/rest-api/dashboard.md index 8c608e36..77524dd8 100644 --- a/docs/dev/rest-api/dashboard.md +++ b/docs/dev/rest-api/dashboard.md @@ -43,10 +43,14 @@ All are `GET`. | Path | Returns | |------|---------| -| `/api/dashboard/status` | Version, `code_hash`, bind host and port, `mode` (`full`, `api_only`, `remote_only`), uptime, `requests_total`, `requests_rejected` (401s), `requests_by_key` (label → count), loaded service count, session count, `max_sessions`, `session_ttl`, job counts by status. | -| `/api/dashboard/sessions` | Every live remote session: `id`, `owner` (username or masked key), `creator_ip`, `service`, `title_id`, `titles`, `tracks`, `title` (the first resolved title's display name), `auth_status`, `auth_error`, `server_account` (the server profile lent to the remote session, else `null`), `client`, `actions`, `created_at` and `last_accessed` (ISO 8601) with `created_ts` and `last_accessed_ts` (Unix epoch, the same clock as log `ts`), `age_seconds`, `idle_seconds`. `client` is whatever the remote client sent as `client` in its session create request (the current CLI sends nothing, so it is `{}`). `actions` is the session's request log, newest last, capped at 500: `ts`, `method`, `action` (`titles`, `tracks`, `segments`, `license`, `prompt`, …), `query`, `status`, `ms`, `bytes_in`, `bytes_out`. | +| `/api/dashboard/status` | Version, `code_hash`, bind host and port, `mode` (`full`, `api_only`, `remote_only`), uptime, `requests_total`, `requests_rejected` (401s), `requests_by_key` (label → count), loaded service count, session count, `max_sessions` (`null` when the operator configured no limit), `session_ttl`, job counts by status. The service count includes services that failed to import - use `/api/dashboard/services` for the number that loaded. | +| `/api/dashboard/sessions` | Every live remote session: `id`, `owner` (username or masked key), `creator_ip`, `service`, `title_id`, `titles`, `tracks`, `title` (the first resolved title's display name), `auth_status`, `auth_error`, `server_account` (the server profile lent to the remote session, else `null`), `client`, `actions`, `created_at` and `last_accessed` (ISO 8601) with `created_ts` and `last_accessed_ts` (Unix epoch, the same clock as log `ts`), `age_seconds`, `idle_seconds`, `log_seq`. `client` is whatever the remote client sent as `client` in its session create request: the CLI sends `version` and `platform`, an older client sends nothing and the field is `{}`. `log_seq` is the last sequence number in the remote session's service log - poll `/api/dashboard/sessions/{id}/logs` when it changes. `actions` is the session's request log, newest last, capped at 500: `ts`, `method`, `action` (`titles`, `tracks`, `segments`, `license`, `prompt`, …), `query`, `status`, `ms`, `bytes_in`, `bytes_out`. | | `/api/dashboard/jobs` | Every download job with full detail, regardless of owner. Empty in `--remote-only` mode. | | `/api/dashboard/logs` | `{"seq": N, "records": [...]}`: the last 1000 log records. `?since=` returns only newer records; `?level=WARNING` sets the minimum level; `?logger=serve` keeps one logger and its children (`aiohttp.access` is the noisiest). `ts` is a Unix epoch in seconds. | +| `/api/dashboard/sessions/{id}/logs` | One remote session's service log: `{"session_id", "last_seq", "records": [...]}`. Each record has `seq`, `ts`, `level`, `message`. `?since=` returns only newer records. 404 when the remote session is unknown. | +| `/api/dashboard/keys` | Every configured API key: what it may do and what it has done. One row per key in `serve.users`, plus `serve.api_secret` when it is set. | +| `/api/dashboard/services` | Every discovered service and its load state, including the ones that failed to import. Not filtered by any allowlist. | +| `/api/dashboard/health` | Preflight: whether this instance could finish a download. Cached 30 seconds. | | `/api/dashboard/events` | Server-Sent Events stream. With `?since=` it returns `{"seq", "stats", "events": [...]}` instead: every event after `seq` (the server keeps the last 20000, log records included, so a poll every few seconds is safe even under a busy access log) as one JSON burst, so a UI that polls every few seconds still sees a remote session that opened and closed in between. | `requests_rejected` counts 401 responses; those never appear in `requests_by_key`. The @@ -64,6 +68,7 @@ those it sends: | `log` | One log record: `seq`, `ts`, `level`, `logger`, `msg`. | | `session` | `action` (`create`, `update` on every request the session makes and when titles or tracks load, or `delete`) plus the full remote session summary, so each event is a complete snapshot. A `delete` also carries `reason`: `closed` (the client ended it), `expired` (TTL sweep) or `evicted` (`max_sessions` reached). A session opened by `--skip-dl` can live only a few seconds, so a UI that wants history must keep its own record of these events. | | `job` | `event` (`queued`, `progress`, `status`, or the terminal status) plus the full job object. | +| `service` | `action` (`staged`, `applied` or `failed`) and `tags`, the services it applies to. A `failed` action carries `errors` and an empty `tags`. Refetch `/api/dashboard/services` when one arrives, instead of polling for a reload. | Every frame carries an `id:` (the event seq). After a dropped connection an `EventSource` reconnects with `Last-Event-ID` and the server replays the events it missed. The connection @@ -80,3 +85,160 @@ es.addEventListener("log", (e) => console.log(JSON.parse(e.data).msg)); CORS is open (`Access-Control-Allow-Origin: *`), so a dashboard page served from another origin works without a proxy. + +## Remote session logs + +A remote session mirrors the service's own `self.log` into a 500-record buffer at `INFO`, +whatever level the server prints at. The remote session summary carries only a truncated +`auth_error`; the buffer holds the real reason a login failed. + +``` +GET /api/dashboard/sessions/{session_id}/logs?since=12 +{"session_id": "...", "last_seq": 14, + "records": [{"seq": 13, "ts": 1756909254.1, "level": "ERROR", "message": "login rejected: MFA required"}]} +``` + +The field name is `records`, the same one `/api/dashboard/logs` uses, so one dashboard reads +both logs the same way. Each record has the shape the client route `/api/session/{id}/logs` +returns. `last_seq` is the buffer's own last sequence number, so an empty page still reports +the true value. + +Reading is a cursor read, not a drain: the client draining the same buffer still gets every +record. It also does not refresh the remote session's idle timer, so polling a remote session here +never keeps it alive or makes it look active. + +!!! note "`log_seq` is a poll trigger, not a push" + A remote session summary reaches the dashboard only when something publishes a `session` + event - a request the remote session makes, or an auth transition. A log line on its own + publishes nothing. Poll this route when `log_seq` changes, and on a timer while a drawer + is open. + +## Keys + +``` +GET /api/dashboard/keys +[{"id": "3ed054be9371", "role": "user", "label": "tier1", "services": ["ATV", "DSNP"], + "server_cdm": false, "server_accounts": false, "server_proxy": false, + "tier": "bot", "rate_limit": 600, "window_used": 412, + "requests": 41300, "rejected": 4, "bytes_out": 103079215104, "last_seen": 1756909188.2}, + {"id": "7c1d0f52a884", "role": "dashboard", "label": "dash…", "services": [], + "server_cdm": false, "server_accounts": false, "server_proxy": false, + "tier": null, "rate_limit": null, "window_used": 0, + "requests": 8140, "rejected": 0, "bytes_out": 41200311, "last_seen": 1756909350.9}] +``` + +`id` is a hash of the API key, so it stays stable across restarts and carries no key material. +Two API keys with no `username` merge into one label in `requests_by_key` but stay separate +here, so use `id` as the identity and `label` only for display. + +| Field | Meaning | +|-------|---------| +| `role` | Which key this is: `user` (a `serve.users` entry), `admin` (`serve.api_secret`) or `dashboard` (`serve.dashboard.key`). Identity, not capability - read the grant fields for what the API key may do. | +| `services` | The effective allowlist, the global list intersected with the API key's. `null` means unrestricted; `[]` means the API key reaches no service route at all. | +| `server_cdm`, `server_accounts` | `false`, `true`, or the list of service tags the grant covers. | +| `server_proxy` | Boolean. Only a literal `true` grants it, and an API key with no `users` entry does not get it. | +| `rate_limit` | Requests per hour, from the API key's own value or its tier's. `null` means no limit. | +| `window_used` | Requests counted in the current hourly window. | +| `rejected` | Responses of 400 or worse that the server attributes to this API key, 429s included. The global `requests_rejected` stays 401-only. | +| `bytes_out` | Response bodies only. An event stream reports no body length, so it counts as zero. | +| `last_seen` | `null` until the API key makes its first request. | + +Only a configured API key gets a row of counters. A request on an open route such as `/api/health` with an unknown key counts as `anonymous`, so an unauthenticated caller cannot grow the table. + +Every API key the server counts has a row here, the dashboard key included, so a +`requests_by_key` bucket other than `anonymous` always matches one. A "traffic by key" panel +reads the rows here, which are unambiguous, and takes `anonymous` from `requests_by_key`. Read +`role` before you draw the bars: the `dashboard` row is the panel measuring its own polling. + +The dashboard key gets `services: []` and no grants unless the operator also gave it a +`serve.users` entry, because it reaches nothing outside `/api/dashboard/`. An API key listed +in more than one place still gets exactly one row. + +### Rate limits + +An API key over its limit gets `429` with a `Retry-After` header: + +```yaml +serve: + tiers: + bot: { rate_limit: 600 } # requests per hour + users: + "": + username: bot-relay + tier: bot + # rate_limit: 900 # optional, wins over the tier +``` + +The server keeps a fixed window, not a sliding one: it opens on the first request and resets +an hour later. An API key with no tier and no override has no limit. A `tier` that names no +entry under `serve.tiers`, or a `rate_limit` that is not a positive whole number, stops the +server at startup, because the alternative is an API key that silently gets no limit. The rate +limit never applies to the dashboard key, and never to `/api/health`. + +## Services + +``` +GET /api/dashboard/services +[{"tag": "MAX", "state": "staged", "error": null, + "commit": "9cc230c98f13b83f6c6fe831595b9702d3ba2786", "staged_commit": "5f0aa17...", + "staged_since": 1756908900.0, "sessions": 1, "jobs": 1, + "aliases": ["HBO"], "geofence": ["US"]}] +``` + +`state` is `loaded`, `staged` or `failed`. A service that failed to import keeps its error +here instead of disappearing, and `/status` still counts it in the service total - count the +`state` values here for the number that loaded. + +A `staged` service has an update on disk that a busy service blocks from the import: the repo +pull already happened, so `commit` (what runs now) and `staged_commit` (what waits) differ. It swaps in when the last job for that tag finishes. Both commits are `null` +for a service from a plain local directory rather than a git repo. + +Watch the `service` event rather than polling for the swap. + +## Health + +``` +GET /api/dashboard/health +{"generated_at": 1756909350.4, "status": "ok", + "checks": [{"id": "ffmpeg", "label": "ffmpeg", "status": "ok", + "detail": "7.1 · /usr/bin/ffmpeg", "ms": 4.1}]} +``` + +An instance can answer every request and still fail every download, because `mp4decrypt` is +missing or a vault's credentials expired. This finds that before a job does. + +`status` is `ok`, `degraded` (an optional dependency is missing) or `failing` (a required one +is). Player binaries are not checked: a headless server never has one and it plays nothing. +Each check is `ok`, `warn` or `fail`. Check ids are the binary names, `cdm`, +`vault:` (the vault's configured `name`, so two vaults of the same type stay apart), and +`proxies`. + +The server caches the result for 30 seconds and keeps every probe shallow: a vault answers one +lookup that should miss, and no probe allocates anything. A failing probe's `detail` has the +vault's password or token masked. This is a panel, not a liveness probe - do not poll it faster +than the cache: every refresh that the cache does not answer costs one real lookup against each +configured vault, which a vault that meters or rate limits its callers will count. + +!!! warning "An unreachable vault costs a slow, partial answer" + The probes run one after another, so a cold read takes as long as the probes that finish + inside the deadline. The endpoint gives up after 15 seconds and answers with the checks that + did complete, plus one `probe` check with the status `fail` naming how many finished. A + database vault that accepts the connection and then stalls is what this deadline is for: its + driver applies no timeout of its own, so without one the request would never answer. + + The probe that timed out keeps running in its thread until its driver returns, and the + server caches the partial answer for the full 30 seconds, so callers cannot stack up more + of them. A second caller waits behind the first, because the server shares one refresh + rather than running it twice. + +!!! warning "Proxy providers are only checked at construction" + The proxy check builds each proxy provider but never asks it for a proxy, because for some + proxy providers that allocates a connection. A proxy provider that only fails on its first + real use still reads `ok`. + +## Counters reset on restart + +Stats, the log ring and the event history are all in memory. A restart drops every number to +zero. `started_at` in the `stats` frame (with `version` and `code_hash`) is the signal: it +arrives in the first frame after a reconnect, so a dashboard can tell a restart from a bug +without any extra field. diff --git a/docs/dev/rest-api/endpoints.md b/docs/dev/rest-api/endpoints.md index 87152715..06c5d61d 100644 --- a/docs/dev/rest-api/endpoints.md +++ b/docs/dev/rest-api/endpoints.md @@ -19,7 +19,7 @@ Start the server with [`unshackle serve`](../../guide/cli-reference.md). By defa - **Service allowlist.** The effective allowlist for your API key filters the service-facing endpoints (the intersection of the global `serve.services` list and your per-key list). The server treats services you cannot use as unknown. !!! warning "Error responses have two shapes" - Structured application errors (raised inside a handler) return `{"status": "error", "error_code": "...", "message": "...", "timestamp": "..."}`. The authentication middleware, however, returns a **different** shape: `{"status": 401, "message": "..."}` with an integer `status` and no `error_code`. Clients must accept both shapes. The [Error responses](#error-responses) section at the bottom of this page gives the full error model. + Structured application errors (raised inside a handler) return `{"status": "error", "error_code": "...", "message": "...", "timestamp": "..."}`. The authentication middleware, however, returns a **different** shape: `{"status": 401, "message": "..."}` with an integer `status` and no `error_code`. The same middleware answers `429` in that shape when an API key goes over its rate limit. Clients must accept both shapes. The [Error responses](#error-responses) section at the bottom of this page gives the full error model. --- @@ -1141,7 +1141,7 @@ The server serialises every structured error raised inside a handler to the same - `debug_info` appears only when you start the server with `--debug-api`. !!! warning "Authentication errors differ" - Errors raised by the API-key middleware use a distinct shape (`{"status": 401, "message": "Secret Key is Empty."}` or `{"status": 401, "message": "Secret Key is Invalid."}`) with an integer `status` and no `error_code`, `timestamp`, or `details`. A malformed JSON body on any POST, by contrast, returns the standard envelope with `error_code: "INVALID_INPUT"`. + Errors raised by the API-key middleware use a distinct shape (`{"status": 401, "message": "Secret Key is Empty."}`, `{"status": 401, "message": "Secret Key is Invalid."}`, or `{"status": 429, "message": "Rate limit exceeded."}` with a `Retry-After` header) with an integer `status` and no `error_code`, `timestamp`, or `details`. A `429` from the middleware therefore carries no `error_code`, unlike the `RATE_LIMITED` code below, which a handler raises for a rate limit the *service* imposed. A malformed JSON body on any POST, by contrast, returns the standard envelope with `error_code: "INVALID_INPUT"`. ### Error codes and default status diff --git a/docs/dev/rest-api/errors.md b/docs/dev/rest-api/errors.md index 9801de69..8a942dcc 100644 --- a/docs/dev/rest-api/errors.md +++ b/docs/dev/rest-api/errors.md @@ -109,6 +109,15 @@ but the defaults hold for the overwhelming majority of responses. | `CONFLICT` | 409 | The target resource is in a state that disallows the action (retrying a running job, prioritising a non-queued job, clearing cache during an active download). | | `RATE_LIMITED` | 429 | The streaming service is rate-limiting requests. Retryable. | +!!! warning "Not every `429` is `RATE_LIMITED`" + The server has a rate limit of its own. An API key over `serve.tiers..rate_limit` + or `serve.users..rate_limit` gets a `429` from the authentication middleware, before + any handler runs, so it carries no `error_code` and is not the `RATE_LIMITED` code above. + It is the shape `{"status": 429, "message": "Rate limit exceeded."}` with a `Retry-After` + header. Read `Retry-After` and wait it out rather than retrying on a backoff of your own: + the window is fixed, so an early retry only spends the next window. See + [Authentication](authentication.md#rate-limit-rejections). + ### Server errors (5xx) | Code | Status | Meaning | diff --git a/docs/reference/configuration/services.md b/docs/reference/configuration/services.md index f4574ef2..098ab166 100644 --- a/docs/reference/configuration/services.md +++ b/docs/reference/configuration/services.md @@ -91,7 +91,7 @@ service the API key has it on, and the local CDM for the others. `true` asks for and falls back to the local CDM when the server refuses; `false` always uses the local CDM. `auth_headers` lists extra header names to send the API key in, tried before the defaults -`X-Secret-Key` and `X-Api-Key`, which are always appended as fallbacks. unshackle sends the +`X-Secret-Key` and `X-Api-Key`, which unshackle always appends as fallbacks. It sends the first name. If the server answers `401`, it retries the same request with the next name, and keeps the one that works for the rest of the HTTP session. Names you give keep your spelling and are not repeated in the fallbacks, so unshackle tries @@ -115,17 +115,18 @@ is the [REST API](../../dev/rest-api/index.md) section. These are the config key | Sub-key | Type | Default | Description | |---------|------|---------|-------------| -| `api_secret` | str | *(unset)* | Master secret accepted in the `X-Secret-Key` header. Required unless the server is started with `--no-key`. | +| `api_secret` | str | *(unset)* | Master secret the server accepts in the `X-Secret-Key` header. Required unless you start the server with `--no-key`. | | `users` | dict | `{}` | Per-user API keys and their allowlists (see below). | +| `tiers` | dict | `{}` | Named settings that `users` entries reference by name (see below). | | `services` | list | *(unset)* | Global service allowlist. Omit to allow all. | | `remote_only` | bool | `false` | Expose only the remote service session endpoints (health, services, search, session) and disable the rest of the REST API. | -| `dashboard` | dict | *(unset)* | Developer dashboard: `key` is the API key for the read-only `/api/dashboard/` endpoints (status, sessions, jobs, logs, SSE events). Unset leaves those routes unregistered. See [dashboard endpoints](../../dev/rest-api/dashboard.md). | +| `dashboard` | dict | *(unset)* | Developer dashboard: `key` is the API key for the read-only `/api/dashboard/` endpoints (status, remote sessions, remote session logs, jobs, keys, services, health, logs, SSE events). Unset leaves those routes unregistered. See [dashboard endpoints](../../dev/rest-api/dashboard.md). | | `session_ttl` | int (s) | `300` | Lifetime of an interactive auth session. | -| `max_sessions` | int | `100` | Maximum concurrent sessions. | +| `max_sessions` | int or null | `100` | Maximum concurrent remote sessions; at the cap the server evicts the oldest. Set it to `null` or `0` for no limit, which is also what `/api/dashboard/status` then reports, so a dashboard can tell "no cap" from a cap that happens to sit at the default. | | `history_limit` | int | `100` | How many finished jobs to retain in history. | | `compression_level` | int | `1` | gzip level for responses. | | `services_refresh_interval` | int (s) | `0` | How often the server pulls the git-backed service repositories in `directories.services` and hot-reloads the services that changed. `0` turns it off. A service with a running or queued job swaps to the new code as soon as its last job finishes. | -| `global_speed_limit` | str | *(unlimited)* | Server-wide download speed cap, e.g. `10M`, `1.5G` or plain bytes/sec (same format as `speed_limit`). One shared budget across all concurrent jobs; per-job speed limits are ignored while it is set. | +| `global_speed_limit` | str | *(unlimited)* | Server-wide download speed cap, e.g. `10M`, `1.5G` or plain bytes/sec (same format as `speed_limit`). One shared budget across all concurrent jobs; the server ignores per-job speed limits while it is set. | | `cdm_overrides` | list or bool | *(unset)* | Allowed per-request CDM overrides: a list of permitted device names, or `true` for any. Unset rejects every override. | | `allow_job_credentials` | bool | `false` | Permit clients to supply credentials per job. | | `server_accounts` | dict | `{}` | Services whose remote sessions use the server's own `credentials` and cookie files instead of the client's (see below). | @@ -146,6 +147,16 @@ client that asks anyway gets a `FORBIDDEN` error. Because a download job always an API key without `server_cdm` for that service also cannot submit or retry `/api/download` jobs. Keys that have no `users` entry, such as `api_secret`, keep server CDM access. +A `tier` names an entry under `serve.tiers`, which holds the settings that several API keys +share. Today a tier carries `rate_limit`, the requests per hour that API key may make; an API +key can also set its own `rate_limit`, which wins over its tier's. An API key with neither has +no limit. Going over the limit answers `429` with a `Retry-After` header. The server keeps a +fixed window rather than a sliding one: it opens on the first request and resets an hour +later. A `tier` that names no entry under `serve.tiers`, or a `rate_limit` that is not a +positive whole number, stops the server at startup, because the alternative is an API key that +silently gets no limit. The rate limit never applies to the dashboard key, and never to +`/api/health`. + `server_accounts` lists, per service, the accounts the server lends to remote sessions. A remote session for a listed service ignores the credentials, cookies, and cache the client sends and uses one of the server's own profiles from `credentials`. For a service that is not listed the @@ -214,9 +225,12 @@ serve: # server-wide download defaults (same keys as the `dl:` block) downloads: 3 best_available: true + tiers: + bot: { rate_limit: 600 } # requests per hour, for every key on this tier users: a1b2c3d4: # this user's API key services: [EXAMPLE1] # may only use EXAMPLE1 + tier: bot # 600 requests/hour; 429 with Retry-After once over e5f6a7b8: server_cdm: true # this key may have the server do the licensing c9d0e1f2: diff --git a/tests/core/test_services_reload.py b/tests/core/test_services_reload.py index a2f0816f..f9096326 100644 --- a/tests/core/test_services_reload.py +++ b/tests/core/test_services_reload.py @@ -72,3 +72,36 @@ def test_busy_service_is_staged_then_applied(service_dir: Path, monkeypatch: pyt assert services.apply_pending(busy=set()) == ["FOO"] assert services.MODULES["FOO"].VERSION == 2 assert services.PENDING == set() + + +def write_broken_service(root: Path, version: int) -> None: + init = root / "FOO" / "__init__.py" + init.write_text(f"raise ImportError('boom {version}')\n") + os.utime(init, (time.time() + version, time.time() + version)) + + +def test_apply_pending_never_reports_a_failed_reimport_as_applied( + service_dir: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """A failed re-import keeps the old module serving, so the tag is not applied.""" + monkeypatch.setattr(services, "PENDING_SINCE", {}) + monkeypatch.setattr(services, "LOADED_COMMITS", {"FOO": "old-head"}) + monkeypatch.setattr(services, "head", lambda dest: "new-head") + (service_dir / ".git").mkdir() + + write_broken_service(service_dir, 2) + monkeypatch.setattr(services, "repo_specs", lambda: ["example/repo"]) + monkeypatch.setattr(services, "refresh_repo", lambda spec: (service_dir, ["~FOO"])) + services.refresh_and_reload(busy={"FOO"}) + assert services.PENDING == {"FOO"} + + assert services.apply_pending(busy=set()) == [] + assert services.MODULES["FOO"].VERSION == 1 + assert services.LOAD_ERRORS and services.LOAD_ERRORS[0].startswith("FOO: ") + # The commit must still name the code that is running, not the one that failed to import. + assert services.LOADED_COMMITS["FOO"] == "old-head" + + write_service(service_dir, 3) + services.PENDING.add("FOO") + assert services.apply_pending(busy=set()) == ["FOO"] + assert services.LOADED_COMMITS["FOO"] == "new-head" diff --git a/tests/remote/unit/test_dashboard.py b/tests/remote/unit/test_dashboard.py index 41dd2197..54391a7f 100644 --- a/tests/remote/unit/test_dashboard.py +++ b/tests/remote/unit/test_dashboard.py @@ -4,15 +4,16 @@ from __future__ import annotations import asyncio import logging +import time import pytest from aiohttp import web from unshackle.core.api.events import bus -from unshackle.core.api.handlers import dashboard_authentication +from unshackle.core.api.handlers import api_key_authentication, dashboard_authentication from unshackle.core.api.routes import cors_middleware, setup_routes from unshackle.core.api.session_store import SessionStore -from unshackle.core.api.stats import RingLogHandler, stats, stats_middleware +from unshackle.core.api.stats import RATE_LIMIT_WINDOW, RingLogHandler, key_id, stats, stats_middleware from unshackle.core.config import config pytestmark = pytest.mark.unit @@ -23,7 +24,8 @@ def dashboard_cfg(monkeypatch): monkeypatch.setitem(config.serve, "dashboard", {"key": "dash-secret"}) monkeypatch.setitem(config.serve, "users", {"tier-key": {"username": "tier1"}}) stats.requests_total = 0 - stats.requests_by_key.clear() + stats.requests_rejected = 0 + stats.keys.clear() def make_app(dashboard: bool = True) -> web.Application: @@ -108,3 +110,414 @@ async def test_session_action_log() -> None: assert queue.get_nowait()["data"]["action"] == "update" and queue.empty() finally: bus.unsubscribe(queue) + + +async def test_key_stats_separate_unnamed_keys_sharing_a_prefix(aiohttp_client, monkeypatch) -> None: + """mask_key truncates to four characters, so two unnamed keys can share a label.""" + monkeypatch.setitem(config.serve, "dashboard", {"key": "dash-secret"}) + monkeypatch.setitem(config.serve, "users", {"abcd-one": {}, "abcd-two": {}}) + stats.keys.clear() + client = await aiohttp_client(make_app()) + await client.get("/api/health", headers={"X-Secret-Key": "abcd-one"}) + for _ in range(2): + await client.get("/api/health", headers={"X-Secret-Key": "abcd-two"}) + + # The legacy label-keyed view still merges them; that is why /keys exists. + assert stats.requests_by_key["abcd…"] == 3 + assert stats.keys[key_id("abcd-one")].requests == 1 + assert stats.keys[key_id("abcd-two")].requests == 2 + + resp = await client.get("/api/dashboard/keys", headers={"X-Secret-Key": "dash-secret"}) + rows = {r["id"]: r for r in await resp.json()} + assert rows[key_id("abcd-one")]["requests"] == 1 + assert rows[key_id("abcd-two")]["requests"] == 2 + assert all(r["rate_limit"] is None for r in rows.values()) + + +def test_rate_limit_window(monkeypatch) -> None: + monkeypatch.setitem(config.serve, "tiers", {"bot": {"rate_limit": 2}}) + monkeypatch.setitem(config.serve, "users", {"k": {"tier": "bot"}, "free": {}}) + stats.keys.clear() + + assert stats.check_rate_limit("k") is None + assert stats.check_rate_limit("k") is None + retry_after = stats.check_rate_limit("k") + assert retry_after is not None and 0 < retry_after <= RATE_LIMIT_WINDOW + + # A key with no tier and no override is unlimited and never opens a window. + assert all(stats.check_rate_limit("free") is None for _ in range(5)) + + # The window is fixed, not sliding: it reopens once the hour has passed. + stats.keys[key_id("k")].window_start -= RATE_LIMIT_WINDOW + 1 + assert stats.check_rate_limit("k") is None + + +def test_rate_limit_per_key_override_beats_tier(monkeypatch) -> None: + monkeypatch.setitem(config.serve, "tiers", {"bot": {"rate_limit": 2}}) + monkeypatch.setitem(config.serve, "users", {"k": {"tier": "bot", "rate_limit": 5}}) + stats.keys.clear() + assert [stats.check_rate_limit("k") for _ in range(5)] == [None] * 5 + assert stats.check_rate_limit("k") is not None + + +async def test_rate_limited_request_returns_429(aiohttp_client, monkeypatch) -> None: + monkeypatch.setitem(config.serve, "dashboard", {"key": "dash-secret"}) + monkeypatch.setitem(config.serve, "tiers", {"bot": {"rate_limit": 1}}) + monkeypatch.setitem(config.serve, "users", {"tier-key": {"username": "tier1", "tier": "bot"}}) + stats.keys.clear() + + app = web.Application( + middlewares=[cors_middleware, stats_middleware, dashboard_authentication, api_key_authentication] + ) + app["config"] = {"users": {"tier-key": {}}} + app["debug_api"] = False + setup_routes(app, remote_only=True, dashboard=True) + client = await aiohttp_client(app) + + assert (await client.get("/api/services", headers={"X-Secret-Key": "tier-key"})).status == 200 + resp = await client.get("/api/services", headers={"X-Secret-Key": "tier-key"}) + assert resp.status == 429 and int(resp.headers["Retry-After"]) > 0 + # The dashboard key leaves the middleware before the limit check, so it always answers. + assert (await client.get("/api/dashboard/status", headers={"X-Secret-Key": "dash-secret"})).status == 200 + + +async def test_dashboard_session_logs(aiohttp_client, dashboard_cfg) -> None: + from unshackle.core.api.session_log import SessionLogBuffer + from unshackle.core.api.session_store import get_session_store + + store = get_session_store() + entry = await store.create("EXAMPLE", object()) + entry.log_buffer = SessionLogBuffer() + entry.log_buffer.append(logging.ERROR, "login rejected: MFA required") + assert entry.summary()["log_seq"] == 1 + + client = await aiohttp_client(make_app()) + headers = {"X-Secret-Key": "dash-secret"} + body = await (await client.get(f"/api/dashboard/sessions/{entry.session_id}/logs", headers=headers)).json() + assert body["last_seq"] == 1 + assert body["records"][0]["message"] == "login rejected: MFA required" + + # A cursor read, so the client draining the same buffer still gets every record. + empty = await (await client.get(f"/api/dashboard/sessions/{entry.session_id}/logs?since=1", headers=headers)).json() + assert empty["records"] == [] and empty["last_seq"] == 1 + + assert (await client.get("/api/dashboard/sessions/nope/logs", headers=headers)).status == 404 + await store.delete(entry.session_id) + + +async def test_dashboard_log_read_does_not_touch_the_session() -> None: + from unshackle.core.api.session_store import SessionStore + + store = SessionStore() + entry = await store.create("EXAMPLE", object()) + before = entry.last_accessed + assert store.peek(entry.session_id) is entry + assert entry.last_accessed == before + # get() is the touching read, and is why the dashboard must not use it. + await store.get(entry.session_id) + assert entry.last_accessed > before + + +async def test_dashboard_services_states(aiohttp_client, dashboard_cfg, monkeypatch) -> None: + from unshackle.core import services as services_module + + tags = services_module.Services.get_tags() + if not tags: + pytest.skip("no services installed") + staged, failed = tags[0], tags[-1] + monkeypatch.setattr(services_module, "PENDING", {staged}) + monkeypatch.setattr(services_module, "PENDING_SINCE", {staged: 1756908900.0}) + monkeypatch.setattr(services_module, "LOAD_ERRORS", [f"{failed}: failed to import - ImportError: boom"]) + + client = await aiohttp_client(make_app()) + rows = { + r["tag"]: r + for r in await (await client.get("/api/dashboard/services", headers={"X-Secret-Key": "dash-secret"})).json() + } + assert rows[failed]["state"] == "failed" and "boom" in rows[failed]["error"] + if staged != failed: + assert rows[staged]["state"] == "staged" and rows[staged]["staged_since"] == 1756908900.0 + assert all(r["state"] == "loaded" for t, r in rows.items() if t not in (staged, failed)) + + +async def test_status_reports_no_session_limit(aiohttp_client, dashboard_cfg, monkeypatch) -> None: + monkeypatch.setitem(config.serve, "max_sessions", None) + client = await aiohttp_client(make_app()) + body = await (await client.get("/api/dashboard/status", headers={"X-Secret-Key": "dash-secret"})).json() + assert body["max_sessions"] is None + + +async def test_dashboard_health(aiohttp_client, dashboard_cfg) -> None: + from unshackle.core.api import handlers + + handlers._health_cache.clear() + client = await aiohttp_client(make_app()) + body = await (await client.get("/api/dashboard/health", headers={"X-Secret-Key": "dash-secret"})).json() + assert body["status"] in {"ok", "degraded", "failing"} + assert {"id", "label", "status", "detail", "ms"} <= set(body["checks"][0]) + assert all(c["status"] in {"ok", "warn", "fail"} for c in body["checks"]) + + +async def test_unknown_key_on_open_route_does_not_grow_key_stats(aiohttp_client, monkeypatch) -> None: + """/api/health answers 200 to any header, so counters keyed by the raw header are a memory leak.""" + monkeypatch.setitem(config.serve, "users", {"real": {}}) + stats.keys.clear() + client = await aiohttp_client(make_app()) + for i in range(20): + assert (await client.get("/api/health", headers={"X-Secret-Key": f"garbage-{i}"})).status == 200 + await client.get("/api/health", headers={"X-Secret-Key": "real"}) + assert set(stats.keys) == {"anonymous", key_id("real")} + assert stats.keys["anonymous"].requests == 20 + + +async def test_refresh_events_never_report_a_failed_reload_as_applied() -> None: + from unshackle.core.api.events import publish_refresh_events + + queue: asyncio.Queue = bus.subscribe() + try: + publish_refresh_events( + [ + { + "spec": "x", + "updated": True, + "changes": ["~GOOD", "~BAD", "~BUSY"], + "deferred": ["BUSY"], + "load_errors": ["BAD: failed to import - ImportError: boom"], + } + ] + ) + events = [] + while not queue.empty(): + events.append(queue.get_nowait()["data"]) + finally: + bus.unsubscribe(queue) + by_action = {e["action"]: e for e in events} + assert by_action["applied"]["tags"] == ["GOOD"] + assert by_action["staged"]["tags"] == ["BUSY"] + assert by_action["failed"]["tags"] == [] and "boom" in by_action["failed"]["errors"][0] + + +def test_health_check_masks_vault_secrets() -> None: + from unshackle.core.api.handlers import health_check + + def probe() -> tuple[str, str]: + raise ConnectionError("could not connect to https://user:hunter2@vault.example/api?token=hunter2") + + detail = health_check("vault:API", "vault API", probe, ["hunter2"])["detail"] + assert "hunter2" not in detail and detail.startswith("ConnectionError") + + +async def test_health_vault_probe_tolerates_api_vault_rejecting_the_probe(aiohttp_server, monkeypatch) -> None: + """An API vault answers code 3/4 to a dummy service or KID; that proves it is up, not down.""" + from unshackle.core.api.handlers import run_health_checks + + async def reject(request: web.Request) -> web.Response: + return web.json_response({"code": 4, "message": "bad kid"}) + + vault_app = web.Application() + vault_app.router.add_get("/{tail:.*}", reject) + server = await aiohttp_server(vault_app) + monkeypatch.setattr( + config, "key_vaults", [{"type": "API", "name": "t", "uri": str(server.make_url("/")), "token": "tok"}] + ) + monkeypatch.setattr(config, "proxy_providers", {}) + checks = {c["id"]: c for c in await asyncio.to_thread(run_health_checks)} + assert checks["vault:t"]["status"] == "ok", checks["vault:t"] + + +def test_health_proxy_probe_reports_a_provider_that_fails_to_build(monkeypatch) -> None: + from unshackle.core.api.handlers import run_health_checks + + monkeypatch.setattr(config, "key_vaults", []) + monkeypatch.setattr(config, "proxy_providers", {"nordvpn": {"username": "u"}}) # missing password + checks = {c["id"]: c for c in run_health_checks()} + assert checks["proxies"]["status"] == "fail" and "TypeError" in checks["proxies"]["detail"] + + +def test_health_proxy_probe_stays_out_of_the_log_ring(monkeypatch, caplog) -> None: + """Rendering a provider fetches its server catalog, and the panel refreshes every 30 s.""" + from unshackle.core.api.handlers import run_health_checks + + monkeypatch.setattr(config, "key_vaults", []) + monkeypatch.setattr(config, "proxy_providers", {"basic": {"us": "http://user:pass@host:8080"}}) + with caplog.at_level(logging.INFO, logger="proxies"): + checks = {c["id"]: c for c in run_health_checks()} + assert checks["proxies"]["status"] == "ok" and "Basic" in checks["proxies"]["detail"] + assert [r.message for r in caplog.records if r.name == "proxies"] == [] + + +async def test_health_vault_checks_are_told_apart_by_name(aiohttp_server, monkeypatch) -> None: + """Two vaults of one type used to answer under the same id and the same label.""" + from unshackle.core.api.handlers import run_health_checks + + async def reject(request: web.Request) -> web.Response: + return web.json_response({"code": 4, "message": "bad kid"}) + + vault_app = web.Application() + vault_app.router.add_get("/{tail:.*}", reject) + server = await aiohttp_server(vault_app) + uri = str(server.make_url("/")) + monkeypatch.setattr( + config, + "key_vaults", + [ + {"type": "API", "name": "primary", "uri": uri, "token": "tok"}, + {"type": "API", "name": "backup", "uri": uri, "token": "tok"}, + ], + ) + monkeypatch.setattr(config, "proxy_providers", {}) + ids = [c["id"] for c in await asyncio.to_thread(run_health_checks) if c["id"].startswith("vault:")] + assert ids == ["vault:primary", "vault:backup"] + + +def test_rate_limit_value_must_be_a_positive_whole_number() -> None: + """key_rate_limit reads anything else as unlimited, so it has to be caught at startup.""" + from unshackle.core.api.stats import key_rate_limit, rate_limit_error + + assert rate_limit_error("serve.tiers.bot", 600) is None + assert rate_limit_error("serve.tiers.bot", None) is None + for bad in ("600", 600.0, 0, -1, True): + assert rate_limit_error("serve.tiers.bot", bad), bad + + with pytest.MonkeyPatch().context() as m: + m.setitem(config.serve, "tiers", {"bot": {"rate_limit": "600"}}) + m.setitem(config.serve, "users", {"k": {"tier": "bot"}}) + assert key_rate_limit("k") is None + + +async def test_dashboard_key_gets_its_own_row(aiohttp_client, dashboard_cfg) -> None: + """configured_key counts the dashboard key, so a row must exist to attribute its polling to.""" + client = await aiohttp_client(make_app()) + headers = {"X-Secret-Key": "dash-secret"} + await client.get("/api/dashboard/status", headers=headers) + rows = await (await client.get("/api/dashboard/keys", headers=headers)).json() + + by_id = {r["id"]: r for r in rows} + assert len(by_id) == len(rows) + dash = by_id[key_id("dash-secret")] + assert dash["role"] == "dashboard" and dash["requests"] > 0 + # null would render as "all services" on a key that reaches no service route. + assert dash["services"] == [] + assert not any((dash["server_cdm"], dash["server_accounts"], dash["server_proxy"])) + assert by_id[key_id("tier-key")]["role"] == "user" + + # Every counted bucket now has a row to attribute it to; only anonymous is left over. + labels = {r["label"] for r in rows} | {"anonymous"} + assert set(stats.requests_by_key) <= labels + + +async def test_dashboard_key_listed_as_a_user_appears_once(aiohttp_client, monkeypatch) -> None: + monkeypatch.setitem(config.serve, "dashboard", {"key": "dash-secret"}) + monkeypatch.setitem(config.serve, "users", {"dash-secret": {"username": "ops", "services": ["EXAMPLE"]}}) + monkeypatch.setitem(config.serve, "api_secret", "dash-secret") + stats.keys.clear() + + client = await aiohttp_client(make_app()) + rows = await (await client.get("/api/dashboard/keys", headers={"X-Secret-Key": "dash-secret"})).json() + assert len(rows) == 1 + # The dashboard role wins as the most specific, and a real users entry keeps its allowlist. + assert rows[0]["role"] == "dashboard" and rows[0]["label"] == "ops" + assert rows[0]["services"] == ["EXAMPLE"] + + +async def test_admin_row_is_distinguishable(aiohttp_client, monkeypatch) -> None: + monkeypatch.setitem(config.serve, "dashboard", {"key": "dash-secret"}) + monkeypatch.setitem(config.serve, "users", {"tier-key": {"username": "tier1"}}) + monkeypatch.setitem(config.serve, "api_secret", "master-secret") + stats.keys.clear() + + client = await aiohttp_client(make_app()) + rows = await (await client.get("/api/dashboard/keys", headers={"X-Secret-Key": "dash-secret"})).json() + roles = {r["role"]: r for r in rows} + assert set(roles) == {"user", "admin", "dashboard"} + # No serve.users entry, so the master secret keeps the implicit access the resolvers give it. + assert roles["admin"]["services"] is None and roles["admin"]["server_cdm"] is True + + +def test_health_check_masks_a_pymysql_password_alias() -> None: + """pymysql accepts passwd and ssl_key_password as well as password; all three reach the driver.""" + from unshackle.core.api.handlers import config_secrets, health_check + + def probe() -> tuple[str, str]: + raise ConnectionError("connect failed for passwd=hunter2 key=keypass1") + + secrets = config_secrets({"type": "MySQL", "host": "db", "passwd": "hunter2", "ssl_key_password": "keypass1"}) + detail = health_check("vault:x", "vault x", probe, secrets)["detail"] + assert "hunter2" not in detail and "keypass1" not in detail + + +def test_health_check_masks_a_nested_config_secret() -> None: + """An API vault's headers and a MySQL vault's ssl are driver arguments with no fixed names.""" + from unshackle.core.api.handlers import config_secrets, health_check + + def probe() -> tuple[str, str]: + raise ConnectionError("rejected header nested-token-1 and cert nested-pass-2") + + secrets = config_secrets( + {"type": "API", "uri": "https://vault.example", "headers": {"X-Api-Key": "nested-token-1"}} + ) + config_secrets({"type": "MySQL", "ssl": {"key_password": "nested-pass-2"}}) + detail = health_check("vault:x", "vault x", probe, secrets)["detail"] + assert "nested-token-1" not in detail and "nested-pass-2" not in detail + # Too short to mask usefully: replacing it would garble the detail without hiding anything. + assert config_secrets({"token": "ab"}) == [] + + +def test_health_check_detail_is_capped_after_redaction() -> None: + """A driver can raise with a whole response body; redaction only hides what it recognises.""" + from unshackle.core.api.handlers import HEALTH_DETAIL_MAX, health_check + + def probe() -> tuple[str, str]: + raise ValueError("API returned an invalid response: " + "hunter2 filler " * 200) + + detail = health_check("vault:x", "vault x", probe, ["hunter2"])["detail"] + assert len(detail) == HEALTH_DETAIL_MAX + 1 and detail.endswith("…") + # Capped after masking, so no secret survives in the part that is kept. + assert "hunter2" not in detail + + +def test_health_proxy_probe_masks_provider_credentials(monkeypatch) -> None: + """The proxy check held only by luck: no provider happens to raise with its own password.""" + from unshackle.core.api.handlers import config_secrets, run_health_checks + + providers = { + "basic": {"us": "http://someone:proxypass1@host:8080"}, + "nordvpn": {"username": "u" * 24, "password": "nordpass1234"}, + } + secrets = config_secrets(providers) + # basic maps a country to a whole proxy URI, so the URI itself is the collected secret. + assert "nordpass1234" in secrets and any("proxypass1" in s for s in secrets) + + monkeypatch.setattr(config, "key_vaults", []) + monkeypatch.setattr(config, "proxy_providers", providers) + detail = {c["id"]: c for c in run_health_checks()}["proxies"]["detail"] + assert "nordpass1234" not in detail and "proxypass1" not in detail + + +async def test_health_probe_timeout_reports_what_finished(aiohttp_client, dashboard_cfg, monkeypatch) -> None: + """A vault that accepts the connection then stalls must not hang the panel forever.""" + from unshackle.core.api import handlers + + def stalling_probe(checks=None): + checks = [] if checks is None else checks + checks.append({"id": "ffmpeg", "label": "ffmpeg", "status": "ok", "detail": "7.1", "ms": 1.0}) + time.sleep(30) # the stalled vault; the deadline must fire long before this returns + checks.append({"id": "never", "label": "never", "status": "ok", "detail": "", "ms": 0.0}) + return checks + + handlers._health_cache.clear() + monkeypatch.setattr(handlers, "run_health_checks", stalling_probe) + monkeypatch.setattr(handlers, "HEALTH_PROBE_TIMEOUT", 0.2) + + client = await aiohttp_client(make_app()) + body = await (await client.get("/api/dashboard/health", headers={"X-Secret-Key": "dash-secret"})).json() + + ids = [c["id"] for c in body["checks"]] + assert "ffmpeg" in ids, "the probe that finished before the deadline must survive" + assert "never" not in ids + assert body["status"] == "failing" + timed_out = next(c for c in body["checks"] if c["id"] == "probe") + assert timed_out["status"] == "fail" and "timed out" in timed_out["detail"] + + # The timeout is cached for the full TTL, so callers cannot stack up more stalled threads. + assert (await client.get("/api/dashboard/health", headers={"X-Secret-Key": "dash-secret"})).status == 200 + assert handlers._health_cache["payload"]["status"] == "failing" diff --git a/tests/remote/unit/test_serve_cli.py b/tests/remote/unit/test_serve_cli.py index 5becf85f..ad287404 100644 --- a/tests/remote/unit/test_serve_cli.py +++ b/tests/remote/unit/test_serve_cli.py @@ -137,3 +137,47 @@ def test_serve_without_no_key_requires_api_secret(runner: CliRunner, monkeypatch result = runner.invoke(serve, ["--api-only"]) assert result.exit_code != 0 assert "api_secret" in (result.output or "").lower() or "api_secret" in str(result.exception).lower() + + +def start_serve(runner: CliRunner, monkeypatch: pytest.MonkeyPatch, serve_cfg: dict) -> tuple[bool, str]: + """Start the command far enough to run the config checks. + + Returns whether it reached ``web.run_app`` and its plain-text output. + """ + from aiohttp import web + + from unshackle.core.config import config as cfg + + started = [] + monkeypatch.setattr(web, "run_app", lambda *a, **kw: started.append(True)) + monkeypatch.setattr(cfg, "serve", dict(serve_cfg)) + result = runner.invoke(serve, ["--api-only", "--no-key"]) + return bool(started), re.sub(r"\x1b\[[0-9;]*m", "", result.output or str(result.exception)) + + +@pytest.mark.parametrize( + "serve_cfg, expected", + [ + ({"tiers": ["bot"]}, "serve.tiers must be a mapping"), + ({"tiers": {"bot": 600}}, "serve.tiers.bot must be a mapping"), + ({"tiers": {"bot": {"rate_limit": 0}}}, "serve.tiers.bot.rate_limit"), + ({"tiers": {"bot": {"rate_limit": True}}}, "serve.tiers.bot.rate_limit"), + ({"users": {"k": {"username": "bot", "rate_limit": -1}}}, "serve.users.bot.rate_limit"), + ({"users": {"k": {"username": "bot", "tier": "gone"}}}, "no such tier 'gone'"), + ], +) +def test_bad_rate_limit_config_stops_the_server( + runner: CliRunner, monkeypatch: pytest.MonkeyPatch, serve_cfg: dict, expected: str +) -> None: + """A tier or rate limit the reader would silently treat as "no limit" must fail at startup.""" + started, output = start_serve(runner, monkeypatch, serve_cfg) + assert not started and expected in output + + +def test_good_rate_limit_config_starts(runner: CliRunner, monkeypatch: pytest.MonkeyPatch) -> None: + started, output = start_serve( + runner, + monkeypatch, + {"tiers": {"bot": {"rate_limit": 600}}, "users": {"k": {"username": "bot", "tier": "bot"}}}, + ) + assert started, output diff --git a/unshackle/commands/serve.py b/unshackle/commands/serve.py index a3a2971b..3ee4b278 100644 --- a/unshackle/commands/serve.py +++ b/unshackle/commands/serve.py @@ -12,15 +12,18 @@ from aiohttp import web from unshackle.core import binaries from unshackle.core.api import cors_middleware, setup_routes, setup_swagger from unshackle.core.api.compression import compression_middleware +from unshackle.core.api.events import publish_refresh_events, publish_service_event from unshackle.core.api.handlers import ( DASHBOARD_PREFIX, + api_key_authentication, dashboard_authentication, dashboard_key, + rate_limit_response, request_secret_key, server_account_regions, validate_server_accounts, ) -from unshackle.core.api.stats import ring, stats, stats_middleware +from unshackle.core.api.stats import key_rate_limit, rate_limit_error, ring, stats, stats_middleware from unshackle.core.config import config from unshackle.core.console import console from unshackle.core.constants import context_settings @@ -33,6 +36,7 @@ def _install_service_refresh(app: web.Application) -> None: log = logging.getLogger("serve") services.log_load_issues() + services.record_loaded_commits() interval = int(config.serve.get("services_refresh_interval", 0) or 0) if interval <= 0 or not services.repo_specs(): return @@ -47,13 +51,16 @@ def _install_service_refresh(app: web.Application) -> None: applied = await asyncio.to_thread(services.apply_pending, manager.busy_services()) if applied: log.info(f"Services reloaded: {', '.join(applied)}") - for r in await asyncio.to_thread(services.refresh_and_reload, manager.busy_services()): + publish_service_event("applied", applied) + repos = await asyncio.to_thread(services.refresh_and_reload, manager.busy_services()) + for r in repos: if r["changes"]: log.info(f"Services refreshed {r['spec']}: {', '.join(r['changes'])}") if r["deferred"]: log.info(f"Services staged until their jobs finish: {', '.join(r['deferred'])}") for err in r["load_errors"]: log.error(f"Service reload failed: {err}") + publish_refresh_events(repos) except Exception: log.exception("Service refresh failed") @@ -219,20 +226,6 @@ def serve( config.serve["playready_devices"] = [] config.serve["playready_devices"].extend(list(config.directories.prds.glob("*.prd"))) - @web.middleware - async def api_key_authentication(request: web.Request, handler) -> web.StreamResponse: - """Authenticate API requests using X-Secret-Key header.""" - if request.path == "/api/health" or request.path.startswith(DASHBOARD_PREFIX): - return await handler(request) - secret_key = request_secret_key(request) - if not secret_key: - return web.json_response({"status": 401, "message": "Secret Key is Empty."}, status=401) - # Constant-time compare against every configured key so a valid key can't be - # recovered byte-by-byte from response timing. - if not any(hmac.compare_digest(secret_key, k) for k in request.app["config"]["users"]): - return web.json_response({"status": 401, "message": "Secret Key is Invalid."}, status=401) - return await handler(request) - remote_only = remote_only or config.serve.get("remote_only", False) if remote_only: api_only = True @@ -268,11 +261,28 @@ def serve( # yaml keys can parse as int; hmac.compare_digest and allowlist lookups need str users = {str(k): v for k, v in users.items()} config.serve["users"] = users + tiers = config.serve.get("tiers") or {} + if not isinstance(tiers, dict): + raise click.ClickException("serve.tiers must be a mapping of tier name to settings") + for tier_name, tier_cfg in tiers.items(): + if not isinstance(tier_cfg, dict): + raise click.ClickException(f"serve.tiers.{tier_name} must be a mapping of setting to value") + if problem := rate_limit_error(f"serve.tiers.{tier_name}", tier_cfg.get("rate_limit")): + raise click.ClickException(problem) for user_key, user_cfg in users.items() if isinstance(users, dict) else []: user_services = user_cfg.get("services") if isinstance(user_cfg, dict) else None + username = user_cfg.get("username", user_key[:8] + "...") if isinstance(user_cfg, dict) else user_key[:8] if user_services: - username = user_cfg.get("username", user_key[:8] + "...") log.info(f"User '{username}' restricted to services: {', '.join(user_services)}") + tier = user_cfg.get("tier") if isinstance(user_cfg, dict) else None + if tier and tier not in tiers: + raise click.ClickException(f"serve.users.{username}.tier: no such tier '{tier}' under serve.tiers") + if isinstance(user_cfg, dict): + if problem := rate_limit_error(f"serve.users.{username}", user_cfg.get("rate_limit")): + raise click.ClickException(problem) + limit = key_rate_limit(user_key) + if limit: + log.info(f"User '{username}' rate limited to {limit} requests/hour") if api_only: log.info("Starting REST API server (pywidevine/pyplayready CDM disabled)") @@ -364,6 +374,14 @@ def serve( secret_key = request_secret_key(request) if request.path.startswith(DASHBOARD_PREFIX): return await handler(request) + if ( + request.path != "/api/health" + and secret_key + and any(hmac.compare_digest(secret_key, k) for k in request.app["config"]["users"]) + ): + limited = rate_limit_response(secret_key) + if limited is not None: + return limited if serve_playready_flag and request.path in ("/playready", "/playready/"): response = await handler(request) elif secret_key and not request.headers.get("X-Secret-Key"): @@ -393,9 +411,9 @@ def serve( app = web.Application( middlewares=[ cors_middleware, + stats_middleware, dashboard_authentication, serve_auth, - stats_middleware, compression_middleware, ] ) diff --git a/unshackle/core/api/download_manager.py b/unshackle/core/api/download_manager.py index 3403f1b2..d90609f0 100644 --- a/unshackle/core/api/download_manager.py +++ b/unshackle/core/api/download_manager.py @@ -15,7 +15,7 @@ from enum import Enum from pathlib import Path from typing import Any, Callable, Dict, List, Optional -from unshackle.core.api.events import bus +from unshackle.core.api.events import bus, publish_service_event from unshackle.core.api.sanitize import sanitize_log from unshackle.core.utils.redact import REDACTED, URL_USERINFO_RE, redact_text @@ -741,6 +741,7 @@ class DownloadQueueManager: applied = await asyncio.to_thread(services.apply_pending, self.busy_services()) if applied: log.info(f"Services reloaded after job completion: {', '.join(applied)}") + publish_service_event("applied", applied) except Exception: log.exception("Applying pending service reloads failed") diff --git a/unshackle/core/api/events.py b/unshackle/core/api/events.py index 6b20fa73..f8011eaa 100644 --- a/unshackle/core/api/events.py +++ b/unshackle/core/api/events.py @@ -54,3 +54,34 @@ def _put_drop(queue: asyncio.Queue, item: Dict[str, Any]) -> None: bus = EventBus() + + +def publish_service_event(action: str, tags: List[str], errors: Optional[List[str]] = None) -> None: + """Tell dashboard listeners that services were staged, applied, or failed to reload. + + Lives here rather than in ``core.services`` so the service loader never imports the API layer. + """ + data: Dict[str, Any] = {"action": action, "tags": list(tags)} + if errors: + data["errors"] = list(errors) + bus.publish("service", data) + + +def publish_refresh_events(repos: List[Dict[str, Any]]) -> None: + """Publish the ``service`` events one ``refresh_and_reload`` result implies. + + A tag counts as applied only when it changed, was not deferred, and did not fail to + import - a failed re-import keeps the old module, so calling it applied would be a lie. + """ + from unshackle.core.services import failed_tags + + for repo in repos: + failed = failed_tags(repo["load_errors"]) + changed = {line[1:] for line in repo["changes"] if line and line[0] in "+~-!"} + applied = sorted(changed - set(repo["deferred"]) - failed) + if applied: + publish_service_event("applied", applied) + if repo["deferred"]: + publish_service_event("staged", repo["deferred"]) + if repo["load_errors"]: + publish_service_event("failed", [], errors=repo["load_errors"]) diff --git a/unshackle/core/api/handlers.py b/unshackle/core/api/handlers.py index d2019825..b4acfdde 100644 --- a/unshackle/core/api/handlers.py +++ b/unshackle/core/api/handlers.py @@ -4,9 +4,11 @@ import enum import json import logging import re +import time +from collections import Counter from datetime import date as date_ from pathlib import Path -from typing import Any, Dict, List, Optional +from typing import Any, Dict, Iterable, List, Optional from aiohttp import web @@ -275,10 +277,10 @@ def setup_list_service( return service_instance -def get_allowed_services(request: Optional[web.Request] = None) -> Optional[List[str]]: - """Get effective service allowlist considering global + per-key config. +def allowed_services_for_key(secret_key: Optional[str]) -> Optional[List[str]]: + """Effective service allowlist for an API key: the global list intersected with the key's. - Returns None if no allowlist restricts the services. + Returns None when nothing restricts the services. """ global_allowed = config.serve.get("services") global_set: Optional[set[str]] = None @@ -286,14 +288,11 @@ def get_allowed_services(request: Optional[web.Request] = None) -> Optional[List global_set = {Services.get_tag(s) for s in global_allowed} key_set: Optional[set[str]] = None - if request: - secret_key = request_secret_key(request) - if secret_key: - users = config.serve.get("users", {}) - user_config = users.get(secret_key, {}) - user_services = user_config.get("services") - if user_services: - key_set = {Services.get_tag(s) for s in user_services} + if secret_key: + users = config.serve.get("users", {}) + user_config = users.get(secret_key, {}) + if isinstance(user_config, dict) and user_config.get("services"): + key_set = {Services.get_tag(s) for s in user_config["services"]} if global_set and key_set: result = global_set & key_set @@ -307,6 +306,11 @@ def get_allowed_services(request: Optional[web.Request] = None) -> Optional[List return list(result) +def get_allowed_services(request: Optional[web.Request] = None) -> Optional[List[str]]: + """The effective service allowlist for a request's API key, or None when unrestricted.""" + return allowed_services_for_key(request_secret_key(request) if request else None) + + def server_cdm_allowed(request: Optional[web.Request] = None, service: Optional[str] = None) -> bool: """Whether the calling API key may have the server operate the CDM licensing for ``service``. @@ -551,6 +555,45 @@ async def dashboard_authentication(request: web.Request, handler: Any) -> web.St return await handler(request) +def rate_limit_response(secret_key: str) -> Optional[web.Response]: + """A 429 when *secret_key* is over its hourly limit, else None after counting the request. + + Call only once the key is known to be valid, so an invalid key cannot burn someone else's + window. Both auth middlewares go through here: the limit has to hold in every server mode. + """ + from unshackle.core.api.stats import stats + + retry_after = stats.check_rate_limit(secret_key) + if retry_after is None: + return None + return web.json_response( + {"status": 429, "message": "Rate limit exceeded."}, + status=429, + headers={"Retry-After": str(retry_after)}, + ) + + +@web.middleware +async def api_key_authentication(request: web.Request, handler: Any) -> web.StreamResponse: + """Gate every non-dashboard route behind a key in ``app["config"]["users"]``, then rate limit it. + + Runs after :func:`dashboard_authentication`, which has already answered the dashboard + routes, so the dashboard key never reaches the limiter. + """ + import hmac + + if request.path == "/api/health" or request.path.startswith(DASHBOARD_PREFIX): + return await handler(request) + secret_key = request_secret_key(request) + if not secret_key: + return web.json_response({"status": 401, "message": "Secret Key is Empty."}, status=401) + # Constant-time compare against every configured key so a valid key can't be + # recovered byte-by-byte from response timing. + if not any(hmac.compare_digest(secret_key, k) for k in request.app["config"]["users"]): + return web.json_response({"status": 401, "message": "Secret Key is Invalid."}, status=401) + return rate_limit_response(secret_key) or await handler(request) + + def caller_key(request: Optional[web.Request] = None) -> str: """The authenticating API key for a request, or 'anonymous' when unauthenticated.""" if not request: @@ -1744,8 +1787,6 @@ async def download_job_events_handler(job_id: str, request: web.Request) -> web. async def dashboard_status_handler(request: web.Request) -> web.Response: """Server-wide stats for the developer dashboard.""" - from collections import Counter - from unshackle.core.api.download_manager import get_download_manager from unshackle.core.api.session_store import get_session_store from unshackle.core.api.stats import stats @@ -1790,6 +1831,296 @@ async def dashboard_logs_handler(request: web.Request) -> web.Response: return web.json_response({"seq": ring.seq, "records": records}) +async def dashboard_session_logs_handler(request: web.Request) -> web.Response: + """One remote session's mirrored service log, for an operator watching a failing auth. + + Reads through ``peek``: a dashboard poll must not refresh the session's idle timer, and + ``SessionLogBuffer.since`` is a cursor read, so this never takes records from the client + draining the same buffer through ``/api/session/{id}/logs``. + """ + from unshackle.core.api.session_store import get_session_store + + session_id = request.match_info["session_id"] + session = get_session_store().peek(session_id) + if session is None: + raise APIError(APIErrorCode.SESSION_NOT_FOUND, f"Remote session not found: {session_id}") + try: + since = int(request.query.get("since", 0)) + except ValueError: + since = 0 + buffer = session.log_buffer + return web.json_response( + { + "session_id": session_id, + "records": buffer.since(since) if buffer else [], + "last_seq": buffer.last_seq if buffer else 0, + } + ) + + +async def dashboard_keys_handler(request: web.Request) -> web.Response: + """Every configured API key: which key it is, what it may do, and what it has done. + + Every key that ``configured_key`` counts gets a row here, so each ``requests_by_key`` + bucket except ``anonymous`` has a row to attribute it to. Without the dashboard key's own + row, a dashboard would render its own polling as traffic from an unknown caller. + """ + from unshackle.core.api.stats import key_id, key_rate_limit, key_tier, mask_key, stats + + users = config.serve.get("users") or {} + api_secret = str(config.serve.get("api_secret") or "") + dashboard = dashboard_key() or "" + + def row(secret_key: str, role: str) -> Dict[str, Any]: + entry = stats.keys.get(key_id(secret_key)) + api_access = role != "dashboard" or secret_key in users + return { + "id": key_id(secret_key), + "role": role, + "label": mask_key(secret_key), + "services": allowed_services_for_key(secret_key) if api_access else [], + "server_cdm": user_grant(secret_key, "server_cdm") if api_access else False, + "server_accounts": user_grant(secret_key, "server_accounts") if api_access else False, + "server_proxy": user_grant(secret_key, "server_proxy") is True if api_access else False, + "tier": key_tier(secret_key), + "rate_limit": key_rate_limit(secret_key), + "window_used": entry.window_used if entry else 0, + "requests": entry.requests if entry else 0, + "rejected": entry.rejected if entry else 0, + "bytes_out": entry.bytes_out if entry else 0, + "last_seen": entry.last_seen if entry else None, + } + + def role_of(secret_key: str) -> str: + """Which key this is. Identity, not capability: the grant fields carry that.""" + if dashboard and secret_key == dashboard: + return "dashboard" + return "admin" if secret_key == api_secret else "user" + + keys = [str(k) for k in users] + for extra in (api_secret, dashboard): + if extra and extra not in keys: + keys.append(extra) + return web.json_response([row(k, role_of(k)) for k in keys]) + + +def user_grant(secret_key: str, name: str) -> Any: + """A per-key grant as configured: False, True, or the list of service tags it covers. + + A key with no ``serve.users`` entry keeps implicit access, matching the resolvers. + """ + user_config = (config.serve.get("users") or {}).get(secret_key) + if not isinstance(user_config, dict): + return name != "server_proxy" + return user_config.get(name, False) + + +async def dashboard_services_handler(request: web.Request) -> web.Response: + """Every discovered service and its load state, including the ones that failed to import. + + Not allowlist-filtered: an operator needs to see the services their keys cannot reach. + """ + from unshackle.core import services as services_module + from unshackle.core.api.download_manager import TERMINAL_STATUSES, get_download_manager + from unshackle.core.api.session_store import get_session_store + from unshackle.core.service_repo import head + + sessions = Counter(entry.service_tag for entry in get_session_store().list()) + jobs = Counter(job.service for job in get_download_manager().list_jobs() if job.status not in TERMINAL_STATUSES) + errors = {err.split(":", 1)[0]: err for err in services_module.LOAD_ERRORS if ":" in err} + staged_commits: Dict[Any, Optional[str]] = {} + + rows = [] + for tag in Services.get_tags(): + error = errors.get(tag) + staged = tag in services_module.PENDING + row: Dict[str, Any] = { + "tag": tag, + "state": "failed" if error else "staged" if staged else "loaded", + "error": error, + "commit": services_module.LOADED_COMMITS.get(tag), + "staged_commit": None, + "staged_since": services_module.PENDING_SINCE.get(tag), + "sessions": sessions.get(tag, 0), + "jobs": jobs.get(tag, 0), + "aliases": list(services_module.ALIASES.get(tag, ())), + "geofence": [], + } + if staged: + # Only staged tags need a git call; the working tree already holds the new commit. + source = services_module.service_source_dir(tag) + if source is not None and source not in staged_commits: + staged_commits[source] = head(source) if (source / ".git").exists() else None + row["staged_commit"] = staged_commits.get(source) + module = services_module.MODULES.get(tag) + if module is not None: + row["geofence"] = list(getattr(module, "GEOFENCE", ()) or ()) + rows.append(row) + return web.json_response(rows) + + +HEALTH_CACHE_TTL = 30.0 +HEALTH_PROBE_KID = "00000000000000000000000000000000" +HEALTH_DETAIL_MAX = 256 +HEALTH_PROBE_TIMEOUT = 15.0 +HEALTH_SECRET_MIN = 4 +SECRET_KEYS = ("password", "passwd", "pwd", "token", "api_key", "apikey", "secret", "auth", "ssl_key_password") +_health_cache: Dict[str, Any] = {} +_health_lock: Optional[asyncio.Lock] = None + + +async def dashboard_health_handler(request: web.Request) -> web.Response: + """Preflight: can this instance actually finish a download? + + A panel, not a liveness probe - the result is cached for 30 s and every probe is shallow, + so reading it never costs the operator a proxy session or a licence. + """ + global _health_lock + if _health_lock is None: + _health_lock = asyncio.Lock() + async with _health_lock: + now = time.time() + cached = _health_cache.get("payload") + if cached and now - _health_cache.get("at", 0.0) < HEALTH_CACHE_TTL: + return web.json_response(cached) + + checks: List[Dict[str, Any]] = [] + try: + await asyncio.wait_for(asyncio.to_thread(run_health_checks, checks), HEALTH_PROBE_TIMEOUT) + except asyncio.TimeoutError: + checks = list(checks) + checks.append( + { + "id": "probe", + "label": "health probe", + "status": "fail", + "detail": f"timed out after {HEALTH_PROBE_TIMEOUT:.0f}s with {len(checks)} checks done", + "ms": round(HEALTH_PROBE_TIMEOUT * 1000, 1), + } + ) + statuses = {c["status"] for c in checks} + payload = { + "generated_at": now, + "status": "failing" if "fail" in statuses else "degraded" if "warn" in statuses else "ok", + "checks": checks, + } + _health_cache.update(at=now, payload=payload) + return web.json_response(payload) + + +def config_secrets(section: Any, nested: bool = False) -> List[str]: + """Values in a vault or proxy config that must never reach a health detail. + + Name-based at the top level, and every scalar below it: a nested mapping is driver + arguments unshackle cannot name in advance (an API vault's ``headers``, a MySQL vault's + ``ssl``, a proxy provider's own block), so it over-collects there on purpose. Short values + are skipped, because masking them would garble the detail without hiding anything. + """ + if isinstance(section, dict): + items: Iterable[tuple[Any, Any]] = section.items() + elif isinstance(section, (list, tuple)): + items = [(None, value) for value in section] + else: + return [str(section)] if nested and section is not None else [] + + found: List[str] = [] + for key, value in items: + if isinstance(value, (dict, list, tuple)): + found += config_secrets(value, nested=True) + elif value is not None and (nested or (isinstance(key, str) and key.lower() in SECRET_KEYS)): + found.append(str(value)) + return [secret for secret in found if len(secret) >= HEALTH_SECRET_MIN] + + +def health_check(check_id: str, label: str, probe: Any, secrets: Iterable[str] = ()) -> Dict[str, Any]: + """Time one health probe and turn any exception into a failed check. + + *secrets* are masked out of the detail: a driver's error string can echo the DSN or URL + it was given. The detail is then capped, because redaction only hides what it recognises + and a driver can raise with a whole response body inside it. Cap after masking, never + before: a cut through a secret would leave the tail of it unreplaced. + """ + started = time.perf_counter() + try: + status, detail = probe() + except Exception as e: + status, detail = "fail", f"{type(e).__name__}: {e}" + masked = redact_text(detail, secrets) or "" + if len(masked) > HEALTH_DETAIL_MAX: + masked = masked[:HEALTH_DETAIL_MAX] + "…" + return { + "id": check_id, + "label": label, + "status": status, + "detail": masked, + "ms": round((time.perf_counter() - started) * 1000, 1), + } + + +def run_health_checks(checks: Optional[List[Dict[str, Any]]] = None) -> List[Dict[str, Any]]: + """Probe binaries, CDM devices, key vaults and proxy providers. Blocking; run in a thread. + + Appends to *checks* as each probe finishes rather than returning only at the end, so a + caller that gives up on the deadline still has the results that did complete. + """ + from unshackle.commands.env import get_dependencies + from unshackle.core.proxies.resolve import initialize_proxy_providers + + if checks is None: + checks = [] + + for dep in (d for d in get_dependencies() if d["cat"] != "Player"): + binary = dep["binary"] + + def probe(binary: Any = binary, dep: Any = dep) -> tuple[str, str]: + if not binary: + return ("fail" if dep["required"] else "warn", "not found on PATH") + version = binary_version(binary) or "unknown version" + return "ok", f"{version} · {binary}" + + checks.append(health_check(dep["name"].lower(), dep["name"], probe)) + + def cdm_probe() -> tuple[str, str]: + wvds = list(config.directories.wvds.glob("*.wvd")) + prds = list(config.directories.prds.glob("*.prd")) + if not wvds and not prds: + return ("fail" if config.cdm else "warn", "no .wvd or .prd device files found") + return "ok", f"{len(wvds)} Widevine, {len(prds)} PlayReady" + + checks.append(health_check("cdm", "CDM devices", cdm_probe)) + + probe_service = next(iter(Services.get_tags()), "health") + for vault_config in config.key_vaults or []: + name = vault_config.get("name") or vault_config.get("type", "vault") + + def vault_probe(vault_config: Any = vault_config) -> tuple[str, str]: + from unshackle.core.vaults import Vaults + + cfg = dict(vault_config) + vaults = Vaults(probe_service) + vaults.load_critical(cfg.pop("type"), **cfg) + vault = vaults.vaults[0] + try: + vault.get_key(HEALTH_PROBE_KID, probe_service) + except Exception as e: + if type(e).__name__ not in ("KeyIdInvalid", "ServiceTagInvalid"): + raise + return "ok", str(vault) + + checks.append(health_check(f"vault:{name}", f"vault {name}", vault_probe, config_secrets(vault_config))) + + def proxy_probe() -> tuple[str, str]: + providers = initialize_proxy_providers(raise_errors=True, quiet=True) + if not providers: + return "ok", "none configured" + return "ok", ", ".join(type(p).__name__ for p in providers) + + if config.proxy_providers: + checks.append(health_check("proxies", "proxy providers", proxy_probe, config_secrets(config.proxy_providers))) + + return checks + + async def dashboard_events_handler(request: web.Request) -> web.StreamResponse: """Server-wide SSE event stream: `log`, `session` and `job` events, plus `stats` every 5 s (also the keep-alive). @@ -2218,11 +2549,13 @@ async def clear_temp_handler(request: Optional[web.Request] = None) -> web.Respo async def refresh_services_handler(request: Optional[web.Request] = None) -> web.Response: """Refresh the service repos configured in directories.services and reload the changed services.""" from unshackle.core.api.download_manager import get_download_manager + from unshackle.core.api.events import publish_refresh_events from unshackle.core.services import refresh_and_reload try: busy = get_download_manager().busy_services() repos = await asyncio.to_thread(refresh_and_reload, busy) + publish_refresh_events(repos) return web.json_response({"refreshed": all(r["updated"] for r in repos), "repos": repos}) except APIError: diff --git a/unshackle/core/api/routes.py b/unshackle/core/api/routes.py index c9bf4bf0..11d6cb7c 100644 --- a/unshackle/core/api/routes.py +++ b/unshackle/core/api/routes.py @@ -21,8 +21,12 @@ from unshackle.core.api.handlers import ( clear_finished_download_jobs_handler, clear_temp_handler, dashboard_events_handler, + dashboard_health_handler, dashboard_jobs_handler, + dashboard_keys_handler, dashboard_logs_handler, + dashboard_services_handler, + dashboard_session_logs_handler, dashboard_sessions_handler, dashboard_status_handler, delete_history_handler, @@ -2123,6 +2127,249 @@ async def dashboard_logs(request: web.Request) -> web.Response: return await dashboard_logs_handler(request) +@api_handler +async def dashboard_session_logs(request: web.Request) -> web.Response: + """ + Dashboard: one remote session's service log. + --- + summary: Dashboard remote session log + description: > + The service's own log output for a remote session, mirrored at INFO regardless of the + server's log level. This is where the real reason for a failed authentication sits, in + full, while the session summary carries only a truncated `auth_error`. + Reading this does not refresh the session's idle timer and does not take records from + the client draining the same buffer through `/api/session/{session_id}/logs`. + Poll when the session summary's `log_seq` changes. + tags: [Dashboard] + parameters: + - name: session_id + in: path + required: true + schema: + type: string + - name: since + in: query + required: false + schema: + type: integer + description: Return only records with seq greater than this value + responses: + '200': + description: Log records plus the buffer's current last_seq + content: + application/json: + schema: + type: object + properties: + session_id: + type: string + last_seq: + type: integer + records: + type: array + items: + type: object + properties: + seq: + type: integer + ts: + type: number + level: + type: string + message: + type: string + '401': + description: Dashboard key missing or invalid + '404': + description: Remote session not found + """ + return await dashboard_session_logs_handler(request) + + +@api_handler +async def dashboard_keys(request: web.Request) -> web.Response: + """ + Dashboard: configured API keys, their grants and their usage. + --- + summary: Dashboard API keys + description: > + Every key in `serve.users`, plus `serve.api_secret` and the dashboard key when they are + configured, with the grants that decide what it may do and the counters for what it has + done. A key listed in more than one of those places still gets exactly one row. + `id` is a hash prefix, stable across restarts and carrying no key material, so two + unnamed keys never merge the way they do in the `requests_by_key` labels. + Every key the server counts has a row here, so a `requests_by_key` bucket other than + `anonymous` always matches one. + `bytes_out` counts response bodies only: an SSE stream reports no content length and + contributes nothing. + tags: [Dashboard] + responses: + '200': + description: One row per configured key + content: + application/json: + schema: + type: array + items: + type: object + properties: + id: + type: string + role: + type: string + enum: [user, admin, dashboard] + description: > + Which key this is, not what it may do: `admin` is `serve.api_secret`, + `dashboard` is `serve.dashboard.key`. The grant fields carry capability. + label: + type: string + services: + type: array + nullable: true + items: + type: string + description: > + Effective allowlist; null when nothing restricts the key, and an empty + list when the key reaches no service route at all, as a dashboard key + with no `serve.users` entry does + server_cdm: + description: false, true, or the list of service tags it covers + server_accounts: + description: false, true, or the list of service tags it covers + server_proxy: + type: boolean + tier: + type: string + nullable: true + rate_limit: + type: integer + nullable: true + description: Requests per hour; null means unlimited + window_used: + type: integer + requests: + type: integer + rejected: + type: integer + bytes_out: + type: integer + last_seen: + type: number + nullable: true + '401': + description: Dashboard key missing or invalid + """ + return await dashboard_keys_handler(request) + + +@api_handler +async def dashboard_services(request: web.Request) -> web.Response: + """ + Dashboard: every discovered service and its load state. + --- + summary: Dashboard services + description: > + Unlike `/api/services` this is not filtered by any allowlist and it keeps the services + that failed to import, with their error. `state` is `loaded`, `staged` (an update is on + disk but a busy service blocks the re-import) or `failed`. + A staged service applies when its last job finishes; watch the `service` event on + `/api/dashboard/events` instead of polling for it. + tags: [Dashboard] + responses: + '200': + description: One row per discovered service + content: + application/json: + schema: + type: array + items: + type: object + properties: + tag: + type: string + state: + type: string + enum: [loaded, staged, failed] + error: + type: string + nullable: true + commit: + type: string + nullable: true + description: Repo HEAD the loaded code was imported from + staged_commit: + type: string + nullable: true + description: Repo HEAD waiting to be imported; only set while staged + staged_since: + type: number + nullable: true + sessions: + type: integer + jobs: + type: integer + aliases: + type: array + items: + type: string + geofence: + type: array + items: + type: string + '401': + description: Dashboard key missing or invalid + """ + return await dashboard_services_handler(request) + + +@api_handler +async def dashboard_health(request: web.Request) -> web.Response: + """ + Dashboard: preflight checks. + --- + summary: Dashboard health preflight + description: > + Whether this instance could actually finish a download: the binaries on PATH, the CDM + device files, each configured key vault and the proxy providers. + A panel, not a liveness probe. The result is cached for 30 seconds and every probe is + shallow, so reading it never spends a proxy session or a licence. A provider that only + fails on first use is therefore not caught here. + tags: [Dashboard] + responses: + '200': + description: Roll-up status plus one entry per check + content: + application/json: + schema: + type: object + properties: + generated_at: + type: number + status: + type: string + enum: [ok, degraded, failing] + checks: + type: array + items: + type: object + properties: + id: + type: string + label: + type: string + status: + type: string + enum: [ok, warn, fail] + detail: + type: string + ms: + type: number + '401': + description: Dashboard key missing or invalid + """ + return await dashboard_health_handler(request) + + @api_handler async def dashboard_events(request: web.Request) -> web.StreamResponse: """ @@ -2201,6 +2448,10 @@ DASHBOARD_ROUTES: list[tuple[str, str, Handler, bool]] = [ ("GET", DASHBOARD_PREFIX + "sessions", dashboard_sessions, True), ("GET", DASHBOARD_PREFIX + "jobs", dashboard_jobs, True), ("GET", DASHBOARD_PREFIX + "logs", dashboard_logs, True), + ("GET", DASHBOARD_PREFIX + "keys", dashboard_keys, True), + ("GET", DASHBOARD_PREFIX + "services", dashboard_services, True), + ("GET", DASHBOARD_PREFIX + "health", dashboard_health, True), + ("GET", DASHBOARD_PREFIX + "sessions/{session_id}/logs", dashboard_session_logs, True), ("GET", DASHBOARD_EVENTS_ROUTE, dashboard_events, True), ] diff --git a/unshackle/core/api/session_store.py b/unshackle/core/api/session_store.py index c751fe2d..2d5b16cd 100644 --- a/unshackle/core/api/session_store.py +++ b/unshackle/core/api/session_store.py @@ -71,6 +71,7 @@ class SessionEntry: "auth_status": self.auth_status.value, "auth_error": self.auth_error, "server_account": self.server_account, + "log_seq": self.log_buffer.last_seq if self.log_buffer else 0, "client": self.client, "actions": list(self.actions), "created_at": self.created_at.isoformat(), @@ -103,9 +104,14 @@ class SessionStore: return config.serve.get("session_ttl", 300) @property - def max_sessions(self) -> int: - """Max concurrent sessions from config.""" - return config.serve.get("max_sessions", 100) + def max_sessions(self) -> Optional[int]: + """Max concurrent sessions from config; None when the operator set no limit. + + ``max_sessions: null`` or ``0`` means unlimited, so a dashboard can tell "no cap" from + a cap that happens to sit at the default. + """ + limit = config.serve.get("max_sessions", 100) + return limit if isinstance(limit, int) and limit > 0 else None async def create( self, @@ -118,9 +124,10 @@ class SessionStore: ) -> SessionEntry: """Make a new remote session with an authenticated service instance.""" async with self._lock: - if len(self._sessions) >= self.max_sessions: + max_sessions = self.max_sessions + if max_sessions is not None and len(self._sessions) >= max_sessions: oldest_id = min(self._sessions, key=lambda k: self._sessions[k].last_accessed) - log.warning(f"Max sessions reached ({self.max_sessions}), evicting oldest: {oldest_id}") + log.warning(f"Max sessions reached ({max_sessions}), evicting oldest: {oldest_id}") _publish("delete", self._sessions.pop(oldest_id), "evicted") session_id = session_id or str(uuid.uuid4()) @@ -154,6 +161,14 @@ class SessionStore: entry.touch() return entry + def peek(self, session_id: str) -> Optional[SessionEntry]: + """A session entry without touching it, for read-only observers. + + ``get`` refreshes ``last_accessed`` and expires stale entries, so an observer polling + through it would keep an idle session alive and always report it as active. + """ + return self._sessions.get(session_id) + async def delete(self, session_id: str) -> bool: """Delete a remote session. Returns True if it existed.""" async with self._lock: diff --git a/unshackle/core/api/stats.py b/unshackle/core/api/stats.py index e80c8867..4dd0a030 100644 --- a/unshackle/core/api/stats.py +++ b/unshackle/core/api/stats.py @@ -2,6 +2,7 @@ from __future__ import annotations +import hashlib import logging import time from collections import Counter, deque @@ -14,17 +15,101 @@ from unshackle.core import __code_hash__, __version__ from unshackle.core.api.events import bus from unshackle.core.config import config +RATE_LIMIT_WINDOW = 3600.0 + + +def _users() -> Dict[str, Any]: + users = (config.serve or {}).get("users") + return users if isinstance(users, dict) else {} + + +def _user_config(key: Optional[str]) -> Optional[Dict[str, Any]]: + user = _users().get(key) if key else None + return user if isinstance(user, dict) else None + + +def configured_key(key: Optional[str]) -> Optional[str]: + """*key* when the config knows it (a user key, the master secret or the dashboard key), else None. + + Open routes answer 200 to any header value, so attributing by the raw header would let an + unauthenticated caller grow ``stats.keys`` without bound. + """ + if not key: + return None + serve = config.serve or {} + dashboard = serve.get("dashboard") + known = {str(k) for k in _users()} | {str(serve.get("api_secret") or "")} + if isinstance(dashboard, dict) and dashboard.get("key"): + known.add(str(dashboard["key"])) + return key if key in known else None + def mask_key(key: Optional[str]) -> str: """A display label for an API key: its configured username, else a truncated prefix.""" if not key: return "anonymous" - user = (config.serve or {}).get("users", {}).get(key) - if isinstance(user, dict) and user.get("username"): + user = _user_config(key) + if user and user.get("username"): return str(user["username"]) return key[:4] + "…" +def key_id(key: Optional[str]) -> str: + """A stable id for an API key. A hash prefix, so two unnamed keys never collide the way + ``mask_key`` does, and the id carries no key material.""" + if not key: + return "anonymous" + return hashlib.sha256(key.encode()).hexdigest()[:12] + + +def key_tier(key: Optional[str]) -> Optional[str]: + """The tier name a key references, when it names one that exists.""" + user = _user_config(key) + tier = user.get("tier") if user else None + return str(tier) if tier else None + + +def key_rate_limit(key: Optional[str]) -> Optional[int]: + """Requests per hour allowed for a key: its own override, else its tier's, else unlimited.""" + user = _user_config(key) + if user is None: + return None + own = user.get("rate_limit") + if isinstance(own, int) and own > 0: + return own + tiers = (config.serve or {}).get("tiers") + tier = tiers.get(user.get("tier")) if isinstance(tiers, dict) and user.get("tier") else None + limit = tier.get("rate_limit") if isinstance(tier, dict) else None + return limit if isinstance(limit, int) and limit > 0 else None + + +def rate_limit_error(where: str, value: Any) -> Optional[str]: + """Why *value* cannot serve as a rate limit, or None when it can. Unset is fine. + + ``key_rate_limit`` reads anything else as "no limit", so a mistyped value would silently + remove the cap the operator wrote down. It lives beside the reader so the two rules cannot + drift apart. + """ + if value is None: + return None + if isinstance(value, bool) or not isinstance(value, int) or value <= 0: + return f"{where}.rate_limit must be a positive whole number of requests per hour, not {value!r}" + return None + + +@dataclass +class KeyStats: + """Per-key counters and the current rate-limit window.""" + + label: str + requests: int = 0 + rejected: int = 0 + bytes_out: int = 0 + last_seen: float = 0.0 + window_start: float = 0.0 + window_used: int = 0 + + @dataclass class ServerStats: started_at: float = field(default_factory=time.time) @@ -33,7 +118,45 @@ class ServerStats: mode: str = "full" requests_total: int = 0 requests_rejected: int = 0 - requests_by_key: Counter = field(default_factory=Counter) + keys: Dict[str, KeyStats] = field(default_factory=dict) + + def key_stats(self, key: Optional[str]) -> KeyStats: + """The counters for an API key, created on first sight.""" + kid = key_id(key) + entry = self.keys.get(kid) + if entry is None: + entry = KeyStats(label=mask_key(key)) + self.keys[kid] = entry + else: + entry.label = mask_key(key) + return entry + + @property + def requests_by_key(self) -> Counter: + """Legacy label-keyed request counts. Two keys sharing a masked label still merge here; + the dashboard reads /api/dashboard/keys for unambiguous per-key figures.""" + counter: Counter = Counter() + for entry in self.keys.values(): + counter[entry.label] += entry.requests + return counter + + def check_rate_limit(self, key: str) -> Optional[int]: + """Seconds to wait when *key* is over its hourly limit, else None after counting the request. + + A fixed window, not a sliding one: cheap, and an operator cap does not need the precision. + """ + limit = key_rate_limit(key) + if limit is None: + return None + entry = self.key_stats(key) + now = time.time() + if now - entry.window_start >= RATE_LIMIT_WINDOW: + entry.window_start = now + entry.window_used = 0 + if entry.window_used >= limit: + return max(1, int(entry.window_start + RATE_LIMIT_WINDOW - now)) + entry.window_used += 1 + return None def to_dict(self) -> Dict[str, Any]: return { @@ -63,7 +186,12 @@ async def stats_middleware(request: web.Request, handler: Any) -> web.StreamResp if response.status == 401: stats.requests_rejected += 1 else: - stats.requests_by_key[mask_key(request_secret_key(request))] += 1 + entry = stats.key_stats(configured_key(request_secret_key(request))) + entry.requests += 1 + entry.last_seen = time.time() + entry.bytes_out += getattr(response, "content_length", None) or 0 + if response.status >= 400: + entry.rejected += 1 session_id = request.match_info.get("session_id") if session_id and request.method != "OPTIONS" and not request.path.endswith("/logs"): from unshackle.core.api.session_store import get_session_store diff --git a/unshackle/core/proxies/resolve.py b/unshackle/core/proxies/resolve.py index dae6f721..5ebe3400 100644 --- a/unshackle/core/proxies/resolve.py +++ b/unshackle/core/proxies/resolve.py @@ -14,8 +14,13 @@ from unshackle.core.utils.redact import mask_proxy log = logging.getLogger("proxies") -def initialize_proxy_providers() -> List[Any]: - """Initialise and return available proxy providers from config.""" +def initialize_proxy_providers(raise_errors: bool = False, quiet: bool = False) -> List[Any]: + """Initialise and return available proxy providers from config. + + A provider that fails to build is logged and skipped, unless *raise_errors* is set. + *quiet* drops the per-provider summary lines: rendering a provider asks some of them for + their server catalog over the network, which a repeated caller must not pay for. + """ proxy_providers: list = [] try: from unshackle.core import binaries @@ -45,13 +50,16 @@ def initialize_proxy_providers() -> List[Any]: if hasattr(binaries, "HolaProxy") and binaries.HolaProxy: proxy_providers.append(Hola()) - for provider in proxy_providers: - log.info(f"Loaded {provider.__class__.__name__}: {provider}") + if not quiet: + for provider in proxy_providers: + log.info(f"Loaded {provider.__class__.__name__}: {provider}") - if not proxy_providers: - log.warning("No proxy providers were loaded. Check your proxy provider configuration in unshackle.yaml") + if not proxy_providers: + log.warning("No proxy providers were loaded. Check your proxy provider configuration in unshackle.yaml") except Exception as e: + if raise_errors: + raise log.warning(f"Failed to initialize some proxy providers: {e}") return proxy_providers diff --git a/unshackle/core/remote_service.py b/unshackle/core/remote_service.py index 51218933..3ca7d630 100644 --- a/unshackle/core/remote_service.py +++ b/unshackle/core/remote_service.py @@ -11,6 +11,7 @@ import base64 import hashlib import logging import re +import sys import time from datetime import date as date_ from enum import Enum @@ -902,6 +903,10 @@ class RemoteService: if cache_data: create_data["cache"] = cache_data + from unshackle.core import __version__ + + create_data["client"] = {"version": __version__, "platform": sys.platform} + result = self.client.post("/api/session/create", create_data) self._session_id = result["session_id"] diff --git a/unshackle/core/services.py b/unshackle/core/services.py index e2ba18f6..ba4b6fc1 100644 --- a/unshackle/core/services.py +++ b/unshackle/core/services.py @@ -5,6 +5,7 @@ import logging import re import sys import threading +import time from pathlib import Path from typing import Any, Iterable @@ -13,7 +14,7 @@ import click from unshackle.core import binaries from unshackle.core.config import config from unshackle.core.service import Service -from unshackle.core.service_repo import DirtyServiceRepo, is_repo_spec, refresh_repo, resolve_service_repo +from unshackle.core.service_repo import DirtyServiceRepo, head, is_repo_spec, refresh_repo, resolve_service_repo from unshackle.core.utilities import import_module_by_path from unshackle.core.utils.redact import redact_path @@ -127,9 +128,42 @@ register_service_binaries(MODULES) ALIASES = {tag: getattr(module, "ALIASES", ()) for tag, module in MODULES.items()} PENDING: set[str] = set() +PENDING_SINCE: dict[str, float] = {} +LOADED_COMMITS: dict[str, str] = {} RELOAD_LOCK = threading.Lock() +def record_loaded_commits(tags: Iterable[str] | None = None) -> None: + """Stamp the repo HEAD each tag was imported from, so a staged update is visibly different. + + ``refresh_and_reload`` pulls before it defers a reload, so the working tree already holds + the new commit while the import is still the old one; without this the two are the same + hash and the staged state cannot be told apart. + """ + wanted = set(tags) if tags is not None else None + by_dir: dict[Path, str | None] = {} + for path in SERVICES: + tag = path.parent.stem + if wanted is not None and tag not in wanted: + continue + source = path.parent.parent + if source not in by_dir: + by_dir[source] = head(source) if (source / ".git").exists() else None + commit = by_dir[source] + if commit: + LOADED_COMMITS[tag] = commit + else: + LOADED_COMMITS.pop(tag, None) + + +def service_source_dir(tag: str) -> Path | None: + """The directory a tag was discovered in: its repo clone, or a local services dir.""" + for path in SERVICES: + if path.parent.stem == tag: + return path.parent.parent + return None + + def reload_services(tags: Iterable[str]) -> list[str]: """Re-import the given service tags after a repo refresh; returns load errors. @@ -151,6 +185,7 @@ def reload_services(tags: Iterable[str]) -> list[str]: SHADOWED[:] = shadowed errors: list[str] = [] + imported: list[str] = [] found = {path.parent.stem: path for path in paths} for tag in wanted: if tag not in found: @@ -160,11 +195,17 @@ def reload_services(tags: Iterable[str]) -> list[str]: try: MODULES[tag] = load_service(found[tag]) ALIASES[tag] = getattr(MODULES[tag], "ALIASES", ()) + imported.append(tag) except Exception as e: errors.append(str(e)) LOAD_ERRORS.extend(errors) register_service_binaries(MODULES) PENDING.difference_update(wanted) + for tag in wanted: + PENDING_SINCE.pop(tag, None) + # Only the tags that imported: a failed re-import keeps the old module running, so + # stamping it with the new HEAD would name code that is not running. + record_loaded_commits(imported) return errors @@ -191,6 +232,9 @@ def refresh_and_reload(busy: set[str] | None = None) -> list[dict[str, Any]]: tags.update(p.parent.stem for p in dest.glob("*/__init__.py")) deferred = sorted(tags & busy) PENDING.update(deferred) + now = time.time() + for tag in deferred: + PENDING_SINCE.setdefault(tag, now) errors = reload_services(tags - busy) repos.append( { @@ -204,13 +248,25 @@ def refresh_and_reload(busy: set[str] | None = None) -> list[dict[str, Any]]: return repos +def failed_tags(errors: Iterable[str]) -> set[str]: + """The tags named by a list of load errors, which are all formatted ``TAG: reason``.""" + return {err.split(":", 1)[0] for err in errors if ":" in err} + + def apply_pending(busy: set[str] | None = None) -> list[str]: - """Reload every staged tag whose service is no longer busy; returns the tags applied.""" + """Reload every staged tag whose service is no longer busy; returns the tags now running. + + A tag whose re-import failed is not one of them: the old module keeps serving, so calling + it applied would tell the dashboard that code is live when it is not. + """ ready = sorted(PENDING - (busy or set())) - if ready: - for err in reload_services(ready): - log.error("Service reload failed: %s", err) - return ready + if not ready: + return [] + errors = reload_services(ready) + for err in errors: + log.error("Service reload failed: %s", err) + failed = failed_tags(errors) + return [tag for tag in ready if tag not in failed] SUMMARY_LOGGED = False