mirror of
https://github.com/unshackle-dl/unshackle.git
synced 2026-11-11 19:45:06 +00:00
feat(serve): add operator dashboard endpoints and per-key rate limits
This commit is contained in:
@@ -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.<name>.rate_limit` or from the API key's own
|
||||
`serve.users.<key>.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.<name>.rate_limit`, overridden by `serve.users.<key>.rate_limit` |
|
||||
| Disable auth | `--no-key` (all requests allowed) |
|
||||
| Config location | `serve.api_secret` and `serve.users` in `unshackle.yaml` |
|
||||
|
||||
|
||||
@@ -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=<seq>` 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=<seq>` 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=<seq>` 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:
|
||||
"<api key>":
|
||||
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:<name>` (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.
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
@@ -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.<name>.rate_limit`
|
||||
or `serve.users.<key>.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 |
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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"
|
||||
|
||||
@@ -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"
|
||||
|
||||
@@ -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
|
||||
|
||||
+36
-18
@@ -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,
|
||||
]
|
||||
)
|
||||
|
||||
@@ -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")
|
||||
|
||||
|
||||
@@ -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"])
|
||||
|
||||
+347
-14
@@ -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:
|
||||
|
||||
@@ -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),
|
||||
]
|
||||
|
||||
|
||||
@@ -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:
|
||||
|
||||
+132
-4
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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"]
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user