Compare commits
6 Commits
c516f659cc
...
main
| Author | SHA1 | Date | |
|---|---|---|---|
| 5c178bb7bd | |||
| 59f0136370 | |||
| e2051f7486 | |||
| 1644999bcb | |||
| 98c80e3a56 | |||
| 3150f78552 |
@@ -100,6 +100,9 @@ HOST=0.0.0.0
|
||||
PORT=8000
|
||||
DEBUG=false
|
||||
LOG_LEVEL=info
|
||||
# Log rendering: "text" (human-readable) or "json" (one object per line, for
|
||||
# Loki/Alloy). The Docker image sets json; text is the default for local dev.
|
||||
LOG_FORMAT=text
|
||||
|
||||
# --- Safety ---
|
||||
# Max simultaneous calls the gateway will place (REST + MCP)
|
||||
|
||||
21
CLAUDE.md
21
CLAUDE.md
@@ -68,7 +68,7 @@ REST / WS / MCP / Dashboard (asyncio, FastAPI)
|
||||
4. **Consistent Patterns** — config via pydantic-settings sub-configs; MCP tools
|
||||
return formatted strings; REST returns Pydantic models; DB access via
|
||||
`session_scope()`. Match the neighbours.
|
||||
5. **Actually Works** — `pytest tests/ -v` (146 tests across 16 files). A change
|
||||
5. **Actually Works** — `pytest tests/ -v` (189 tests across 19 files). A change
|
||||
to call placement, routing, the classifier, or auth needs a test. The suite
|
||||
runs against SQLite (`aiosqlite`) and the mock SIP engine — no trunk, no
|
||||
Postgres required to test.
|
||||
@@ -203,15 +203,24 @@ Surfaced honestly so you don't rediscover them as surprises. The README's
|
||||
Phase 4/5/6 checklists track most of these; **don't fold fixes into unrelated
|
||||
work** — raise them.
|
||||
|
||||
- **No structured JSON logging.** Logging is plain `logging.basicConfig` in
|
||||
[main.py](main.py); there's no `LOG_FORMAT`/JSON path (README Phase 4 has this
|
||||
unchecked). If Heluca observability wants JSON logs shipped to a collector,
|
||||
that's a deliberate piece of work, not a drive-by.
|
||||
- ~~**No structured JSON logging.**~~ Done: `LOG_FORMAT=json` in
|
||||
[core/logging_config.py](core/logging_config.py), applied at import and again
|
||||
in `lifespan` because uvicorn installs its own handlers (`propagate=False`)
|
||||
after importing the app. The access log is included, with `status_code` as a
|
||||
number so Loki can range-filter it. Text remains the default; the Docker image
|
||||
sets json.
|
||||
- **No `/metrics` endpoint and no Prometheus.** Unlike the metrics-bearing
|
||||
estate services, there's no exposition endpoint here yet.
|
||||
- **No health-probe access-log filter.** Every `/health` poll hits the access
|
||||
log. Other estate services suppress probe noise; this one doesn't.
|
||||
- **No rate limiting** on API endpoints (README Phase 4, unchecked).
|
||||
- ~~**No rate limiting** on API endpoints.~~ Done, but *narrowly*: only the
|
||||
unauthenticated `/auth/*` routes are limited
|
||||
([core/rate_limit.py](core/rate_limit.py)), because every other surface is
|
||||
already owner-gated and a limit there would throttle the sole operator. The
|
||||
limiter keys on the **socket peer, not `X-Forwarded-For`** — behind the
|
||||
estate's reverse proxy that means per-proxy, not per-caller. Per-caller limits
|
||||
need an explicit trusted-proxy config; don't silently start trusting the
|
||||
header.
|
||||
- **Docker: single-image `Dockerfile` + `docker-compose.yaml`** (app +
|
||||
`postgres:17`) ship in-repo; the Gitea CI (`cve-scan-docker-build.yml`) builds
|
||||
the image on push to `main`. The compose stack requires SSO enabled
|
||||
|
||||
@@ -45,6 +45,11 @@ RUN pip install --no-cache-dir -e . \
|
||||
|
||||
EXPOSE 21081
|
||||
|
||||
# Structured logs by default in the container: the host's Alloy agent reads
|
||||
# stdout and ships it to Loki, where text lines arrive as an unqueryable blob.
|
||||
# Overridable (LOG_FORMAT=text) for interactive `docker run` debugging.
|
||||
ENV LOG_FORMAT=json
|
||||
|
||||
# Migrations run in the app's own init_db() on boot (db/database.py), so no
|
||||
# separate `alembic upgrade` here. Bind host/port from the same env vars
|
||||
# pydantic-settings reads (HOST/PORT) so configured values and the actual bind
|
||||
|
||||
48
README.md
48
README.md
@@ -52,8 +52,9 @@ You give it a phone number and an intent ("dispute a charge on my December state
|
||||
## What's Implemented
|
||||
|
||||
### Core Engine
|
||||
- **Sippy B2BUA Engine** (`core/sippy_engine.py`) — SIP call control, DTMF, bridging, conference, trunk registration
|
||||
- **PJSUA2 Media Pipeline** (`core/media_pipeline.py`) — Audio routing, recording ports, conference bridge, WAV playback (stub mode until the `pjsua2` bindings are installed — see note below)
|
||||
- **Sippy B2BUA Engine** (`core/sippy_engine.py`) — SIP call control, DTMF, bridging, conference, trunk registration. Signalling only: no audio reaches the classifier on this path
|
||||
- **PJSUA2 SIP Engine** (`core/pjsua_engine.py`) — Places the call itself so it owns the dialog, which is the only way PJSUA2 will surface RTP. Select with `SIP_ENGINE=pjsua2`; see [docs/architecture.md](docs/architecture.md#media-plane-why-pjsua2-places-the-call)
|
||||
- **PJSUA2 Media Pipeline** (`core/media_pipeline.py`) — Audio routing, capture ports, conference bridge, WAV playback (stub mode until the `pjsua2` bindings are installed — see note below)
|
||||
- **Call Manager** (`core/call_manager.py`) — Active call state tracking, lifecycle management
|
||||
- **Event Bus** (`core/event_bus.py`) — Async pub/sub with per-subscriber queues, type filtering, history
|
||||
|
||||
@@ -99,8 +100,13 @@ hold-slayer/
|
||||
├── config.py # Pydantic settings from .env
|
||||
├── core/
|
||||
│ ├── gateway.py # Top-level gateway orchestrator
|
||||
│ ├── sippy_engine.py # Sippy B2BUA SIP engine
|
||||
│ ├── dial_plan.py # Emergency-number guard + number normalisation
|
||||
│ ├── sip_engine.py # SIPEngine ABC + MockSIPEngine
|
||||
│ ├── sippy_engine.py # Sippy B2BUA SIP engine (signalling only)
|
||||
│ ├── pjsua_engine.py # PJSUA2 SIP engine (call control + media)
|
||||
│ ├── media_pipeline.py # PJSUA2 audio routing
|
||||
│ ├── logging_config.py # Text/JSON log formatting
|
||||
│ ├── rate_limit.py # Fixed-window limiter for the /auth/* edge
|
||||
│ ├── call_manager.py # Active call state management
|
||||
│ └── event_bus.py # Async pub/sub event bus
|
||||
├── services/
|
||||
@@ -213,6 +219,15 @@ uvicorn main:app --host 0.0.0.0 --port 8000
|
||||
pytest tests/ -v
|
||||
```
|
||||
|
||||
No external services required — the suite runs against SQLite (`aiosqlite`) and
|
||||
the mock SIP engine, so it needs neither a trunk nor PostgreSQL.
|
||||
|
||||
For the parts a unit test cannot reach — real SIP signalling, RTP, IVR
|
||||
navigation against a live switch — there is an **Asterisk lab**: a fake PSTN
|
||||
that answers calls, plays hold music, and runs scripted IVR menus, so call
|
||||
paths can be exercised without dialling a real number or incurring telephony
|
||||
charges. See [tests/lab/README.md](tests/lab/README.md).
|
||||
|
||||
## Docker
|
||||
|
||||
A single image bundles the FastAPI process and the built dashboard (the node
|
||||
@@ -373,11 +388,19 @@ All configuration is via environment variables (see `.env.example`):
|
||||
| `OWNER_NAME` | Casdoor username of the single operator (owner) | — (required if SSO on) |
|
||||
| `PUBLIC_BASE_URL` | Public base URL for OAuth discovery (else derived from headers) | — |
|
||||
| `MAX_CONCURRENT_CALLS` | Cap on simultaneous outbound calls | `4` |
|
||||
| `HOST` | Bind address (off-loopback requires SSO enabled) | `0.0.0.0` |
|
||||
| `PORT` | Bind port | `8000` |
|
||||
| `DEBUG` | SQLAlchemy echo + uvicorn reload | `false` |
|
||||
| `LOG_LEVEL` | Root log level (`debug`/`info`/`warning`/`error`) | `info` |
|
||||
| `LOG_FORMAT` | `text` (human-readable) or `json` (structured, for Loki) | `text` |
|
||||
| `USE_MOCK_SIP` | Run the mock SIP engine — no real calls. Must be asked for | `false` |
|
||||
| `SIP_ENGINE` | `sippy` (signalling only) or `pjsua2` (call control + media) | `sippy` |
|
||||
| `NOTIFY_SMS_NUMBER` | SMS notification number (optional) | — |
|
||||
| `SIP_TRUNK_HOST` | Your SIP provider hostname | — |
|
||||
| `SIP_TRUNK_USERNAME` | SIP auth username | — |
|
||||
| `SIP_TRUNK_PASSWORD` | SIP auth password | — |
|
||||
| `SIP_TRUNK_DID` | Your phone number (E.164) | — |
|
||||
| `GATEWAY_SIP_PORT` | Port for device registration | `5080` |
|
||||
| `GATEWAY_SIP_PORT` | Port for device registration | `5060` |
|
||||
| `SPEACHES_URL` | Speaches/Whisper STT endpoint | `http://localhost:22070` |
|
||||
| `LLM_BASE_URL` | OpenAI-compatible LLM endpoint | `http://localhost:11434/v1` |
|
||||
| `LLM_MODEL` | Model name for IVR analysis | `llama3` |
|
||||
@@ -391,11 +414,11 @@ All configuration is via environment variables (see `.env.example`):
|
||||
|
||||
## Tech Stack
|
||||
|
||||
- **Python 3.12+** + **asyncio** — Single-process async architecture
|
||||
- **Python 3.12+** + **asyncio** — Single process. Not single-threaded: the SIP stacks run their own event loops on separate OS threads, crossed only through defined funnels ([docs/architecture.md](docs/architecture.md#threading-model))
|
||||
- **FastAPI** — REST API + WebSocket server
|
||||
- **SvelteKit** — Dashboard UI (built static, served by FastAPI at `/`)
|
||||
- **Sippy B2BUA** — SIP call control and DTMF
|
||||
- **PJSUA2** — Media pipeline, conference bridge, recording, WAV playback
|
||||
- **Sippy B2BUA** — SIP call control and DTMF (`SIP_ENGINE=sippy`, signalling only)
|
||||
- **PJSUA2** — Call control + media: conference bridge, capture ports, recording, WAV playback (`SIP_ENGINE=pjsua2`)
|
||||
- **Speaches** (Whisper) — Speech-to-text
|
||||
- **Rhema** (Kokoro) — Text-to-speech (OpenAI-compatible `/v1/audio/speech`)
|
||||
- **Ollama / vLLM / OpenAI** — LLM for IVR menu analysis and receptionist intent capture
|
||||
@@ -408,6 +431,9 @@ Full documentation is in [`/docs`](docs/README.md):
|
||||
|
||||
- [Architecture](docs/architecture.md) — System design, data flow, threading model
|
||||
- [Core Engine](docs/core-engine.md) — SIP engine, media pipeline, call manager, event bus
|
||||
- [Dial Plan](docs/dial-plan.md) — Number normalisation and the emergency-number guard
|
||||
- [PJSUA2 Build](docs/pjsua2-build.md) — Building the bindings (not pip-installable)
|
||||
- [Asterisk Lab](tests/lab/README.md) — The fake PSTN used for media validation
|
||||
- [Hold Slayer Service](docs/hold-slayer-service.md) — IVR navigation, hold detection, human detection
|
||||
- [Audio Classifier](docs/audio-classifier.md) — Waveform analysis, feature extraction, classification
|
||||
- [Services](docs/services.md) — LLM client, transcription, recording, analytics, notifications
|
||||
@@ -443,15 +469,15 @@ Full documentation is in [`/docs`](docs/README.md):
|
||||
- [x] Notification service (WebSocket + SMS)
|
||||
- [x] Service wiring in main.py lifespan
|
||||
|
||||
### Phase 4: Production Hardening 🚧
|
||||
### Phase 4: Production Hardening ✅
|
||||
|
||||
- [x] Alembic database migrations (baseline + upgrade-on-boot)
|
||||
- [x] API authentication — Casdoor SSO (browser JWT) + owner-minted PATs, owner-only across REST/WS/MCP
|
||||
- [x] Emergency-number guard + concurrent-call cap on outbound calls
|
||||
- [ ] Rate limiting on API endpoints
|
||||
- [ ] Structured JSON logging
|
||||
- [x] Rate limiting on the unauthenticated `/auth/*` edge (everything else is owner-gated; see [core/rate_limit.py](core/rate_limit.py))
|
||||
- [x] Structured JSON logging (`LOG_FORMAT=json`, uvicorn access log included)
|
||||
- [x] Honest /health — engine mode, DB ping, trunk registration, STT/TTS availability
|
||||
- [ ] Graceful degradation (classifier works without STT, etc.)
|
||||
- [x] Graceful degradation — a down STT/LLM/TTS degrades the call and publishes an `ERROR` event naming the service, rather than aborting it
|
||||
- [x] Docker Compose (Hold Slayer + PostgreSQL)
|
||||
|
||||
### Phase 5: Additional Services 🚧
|
||||
|
||||
22
api/auth.py
22
api/auth.py
@@ -15,15 +15,25 @@ non-owner sees an "access denied" screen instead of a bare 401.
|
||||
import secrets
|
||||
from urllib.parse import urlencode
|
||||
|
||||
from fastapi import APIRouter, HTTPException, Query, Request
|
||||
from fastapi import APIRouter, Depends, HTTPException, Query, Request
|
||||
from fastapi.responses import HTMLResponse, JSONResponse, RedirectResponse
|
||||
|
||||
from auth import get_sdk, is_owner, resolve_from_header_or_query
|
||||
from config import get_settings
|
||||
from core.rate_limit import rate_limit
|
||||
from db.database import session_scope
|
||||
|
||||
router = APIRouter(prefix="/auth", tags=["auth"])
|
||||
|
||||
# These are the only routes that must answer before an identity exists, so they
|
||||
# are the only ones worth limiting — everything else is already owner-gated.
|
||||
# `/callback` and `/refresh-callback` each trigger an outbound token exchange
|
||||
# with Casdoor, and `/me` opens a DB session per call; all three are
|
||||
# unauthenticated work an attacker controls. See core/rate_limit.py.
|
||||
_limit_callback = [Depends(rate_limit("auth:callback", limit=10))]
|
||||
_limit_me = [Depends(rate_limit("auth:me"))]
|
||||
_limit_redirect = [Depends(rate_limit("auth:redirect"))]
|
||||
|
||||
|
||||
def _build_casdoor_auth_url(
|
||||
callback: str,
|
||||
@@ -50,7 +60,7 @@ def _build_casdoor_auth_url(
|
||||
return f"{c.endpoint.rstrip('/')}/login/oauth/authorize?{urlencode(params)}"
|
||||
|
||||
|
||||
@router.get("/login")
|
||||
@router.get("/login", dependencies=_limit_redirect)
|
||||
async def login(request: Request, redirect_uri: str = Query(None)):
|
||||
"""Redirect the browser to the Casdoor authorization page.
|
||||
|
||||
@@ -64,7 +74,7 @@ async def login(request: Request, redirect_uri: str = Query(None)):
|
||||
return RedirectResponse(url=_build_casdoor_auth_url(callback))
|
||||
|
||||
|
||||
@router.get("/callback")
|
||||
@router.get("/callback", dependencies=_limit_callback)
|
||||
async def callback(
|
||||
code: str = Query(...),
|
||||
state: str = Query(None),
|
||||
@@ -89,7 +99,7 @@ async def callback(
|
||||
return RedirectResponse(url=f"/#token={access_token}")
|
||||
|
||||
|
||||
@router.get("/silent-refresh")
|
||||
@router.get("/silent-refresh", dependencies=_limit_redirect)
|
||||
async def silent_refresh(request: Request):
|
||||
"""Start a silent token refresh via hidden iframe (``prompt=none``).
|
||||
|
||||
@@ -104,7 +114,7 @@ async def silent_refresh(request: Request):
|
||||
return RedirectResponse(url=_build_casdoor_auth_url(callback, prompt="none"))
|
||||
|
||||
|
||||
@router.get("/refresh-callback")
|
||||
@router.get("/refresh-callback", dependencies=_limit_callback)
|
||||
async def refresh_callback(
|
||||
code: str = Query(None),
|
||||
error: str = Query(None),
|
||||
@@ -140,7 +150,7 @@ async def refresh_callback(
|
||||
)
|
||||
|
||||
|
||||
@router.get("/me")
|
||||
@router.get("/me", dependencies=_limit_me)
|
||||
async def me(request: Request):
|
||||
"""Return the current authenticated user's profile + ``is_owner``.
|
||||
|
||||
|
||||
@@ -145,6 +145,11 @@ class Settings(BaseSettings):
|
||||
debug: bool = False
|
||||
log_level: str = "info"
|
||||
|
||||
# Log rendering: "text" (human-readable, for a terminal) or "json" (one
|
||||
# object per line, for Loki). Text is the default so local dev is readable;
|
||||
# the container sets LOG_FORMAT=json. See core/logging_config.py.
|
||||
log_format: str = "text"
|
||||
|
||||
# Auth — Casdoor SSO for the browser + owner-minted PATs for MCP/CLI,
|
||||
# gated to a single owner. `owner_name` is the Casdoor username that owns
|
||||
# this gateway (everyone else gets 403). `public_base_url` seeds the OAuth
|
||||
|
||||
175
core/logging_config.py
Normal file
175
core/logging_config.py
Normal file
@@ -0,0 +1,175 @@
|
||||
"""
|
||||
Logging configuration — human-readable text or structured JSON.
|
||||
|
||||
Hold Slayer's logs are shipped to Loki by the host's Alloy agent, which reads
|
||||
the container's stdout. Text logs arrive there as an opaque blob: filtering on
|
||||
a status code or a call ID means regex over a formatted string. JSON lines
|
||||
arrive as queryable fields.
|
||||
|
||||
Two things about this are less obvious than they look, and both are the reason
|
||||
this module exists instead of a `format=` argument on `basicConfig`:
|
||||
|
||||
1. **Uvicorn brings its own handlers.** `uvicorn.config.LOGGING_CONFIG` attaches
|
||||
a `StreamHandler` to `uvicorn` and `uvicorn.access` with `propagate: False`,
|
||||
so those records never reach the root logger's formatter. Configuring only
|
||||
the root would leave the access log — the highest-volume, most useful stream
|
||||
— as plain colourised text next to our JSON. `configure_logging` reaches into
|
||||
those two loggers explicitly.
|
||||
|
||||
2. **The access record's payload is in `record.args`, not the message.** Uvicorn
|
||||
logs access lines as a 5-tuple `(client_addr, method, full_path, http_version,
|
||||
status_code)` and lets its `AccessFormatter` interpolate them. Formatting the
|
||||
message would throw that structure away and force Loki to parse it back out,
|
||||
so `JSONFormatter` unpacks the tuple into real fields.
|
||||
|
||||
Text mode stays the default: it is what a developer wants on a terminal, and a
|
||||
JSON-only logger makes local debugging worse. Production opts in via
|
||||
`LOG_FORMAT=json`.
|
||||
"""
|
||||
|
||||
import datetime as _dt
|
||||
import json
|
||||
import logging
|
||||
import sys
|
||||
from typing import Any
|
||||
|
||||
# LogRecord attributes that are either already represented in our output or are
|
||||
# formatting machinery. Anything on a record that is *not* here is treated as a
|
||||
# caller-supplied `extra=` field and promoted into the JSON object.
|
||||
_RESERVED = frozenset(
|
||||
{
|
||||
"args",
|
||||
"asctime",
|
||||
# Uvicorn passes an ANSI-colourised duplicate of the message as
|
||||
# `extra={"color_message": ...}` for its own formatter to prefer. Left
|
||||
# unfiltered, generic extra-promotion copies escape codes into every
|
||||
# startup line — the same unreadable-in-Grafana problem as the lab's
|
||||
# Asterisk logs.
|
||||
"color_message",
|
||||
"created",
|
||||
"exc_info",
|
||||
"exc_text",
|
||||
"filename",
|
||||
"funcName",
|
||||
"levelname",
|
||||
"levelno",
|
||||
"lineno",
|
||||
"module",
|
||||
"msecs",
|
||||
"message",
|
||||
"msg",
|
||||
"name",
|
||||
"pathname",
|
||||
"process",
|
||||
"processName",
|
||||
"relativeCreated",
|
||||
"stack_info",
|
||||
"taskName",
|
||||
"thread",
|
||||
"threadName",
|
||||
}
|
||||
)
|
||||
|
||||
# Uvicorn's own access-log tuple, in order.
|
||||
_ACCESS_FIELDS = ("client_addr", "method", "path", "http_version", "status_code")
|
||||
|
||||
TEXT_FORMAT = "%(asctime)s | %(levelname)-7s | %(name)s | %(message)s"
|
||||
|
||||
|
||||
class JSONFormatter(logging.Formatter):
|
||||
"""Render a LogRecord as a single-line JSON object."""
|
||||
|
||||
def format(self, record: logging.LogRecord) -> str:
|
||||
payload: dict[str, Any] = {
|
||||
# RFC 3339 in UTC. `logging`'s default asctime is local-time and
|
||||
# date-less, which makes correlating with Loki's own timestamps
|
||||
# unnecessarily hard.
|
||||
"ts": _dt.datetime.fromtimestamp(record.created, tz=_dt.UTC).isoformat(
|
||||
timespec="milliseconds"
|
||||
),
|
||||
"level": record.levelname,
|
||||
"logger": record.name,
|
||||
}
|
||||
|
||||
if record.name == "uvicorn.access" and isinstance(record.args, tuple):
|
||||
payload.update(_access_fields(record))
|
||||
else:
|
||||
payload["msg"] = record.getMessage()
|
||||
|
||||
# Thread name matters here in a way it doesn't in a single-context app:
|
||||
# this process runs the asyncio loop, the Sippy ED thread, and PJSUA2
|
||||
# worker threads, and "which context logged this" is usually the first
|
||||
# question when debugging a call.
|
||||
if record.threadName and record.threadName != "MainThread":
|
||||
payload["thread"] = record.threadName
|
||||
|
||||
if record.exc_info:
|
||||
payload["exc"] = self.formatException(record.exc_info)
|
||||
if record.stack_info:
|
||||
payload["stack"] = self.formatStack(record.stack_info)
|
||||
|
||||
for key, value in record.__dict__.items():
|
||||
if key not in _RESERVED and not key.startswith("_"):
|
||||
payload[key] = _safe(value)
|
||||
|
||||
return json.dumps(payload, default=str, separators=(",", ":"))
|
||||
|
||||
|
||||
def _access_fields(record: logging.LogRecord) -> dict[str, Any]:
|
||||
"""Unpack uvicorn's access-log arg tuple into named fields.
|
||||
|
||||
Falls back to the interpolated message if uvicorn ever changes the tuple's
|
||||
shape — a log line with a slightly wrong shape beats an exception inside the
|
||||
logging path taking out the request.
|
||||
"""
|
||||
args = record.args
|
||||
if not isinstance(args, tuple) or len(args) != len(_ACCESS_FIELDS):
|
||||
return {"msg": record.getMessage()}
|
||||
|
||||
fields: dict[str, Any] = dict(zip(_ACCESS_FIELDS, args))
|
||||
try:
|
||||
fields["status_code"] = int(fields["status_code"])
|
||||
except (TypeError, ValueError):
|
||||
pass
|
||||
return fields
|
||||
|
||||
|
||||
def _safe(value: Any) -> Any:
|
||||
"""Keep JSON-native types; stringify everything else."""
|
||||
if isinstance(value, (str, int, float, bool, type(None))):
|
||||
return value
|
||||
return str(value)
|
||||
|
||||
|
||||
def configure_logging(log_format: str, log_level: str) -> None:
|
||||
"""Install the root and uvicorn log handlers.
|
||||
|
||||
Idempotent: existing root handlers are removed first, so calling this after
|
||||
uvicorn has configured itself replaces its formatting rather than adding a
|
||||
second stream (which is how you get every line twice).
|
||||
"""
|
||||
level = getattr(logging, log_level.upper(), logging.INFO)
|
||||
use_json = log_format.lower() == "json"
|
||||
|
||||
formatter: logging.Formatter = (
|
||||
JSONFormatter() if use_json else logging.Formatter(TEXT_FORMAT, datefmt="%H:%M:%S")
|
||||
)
|
||||
|
||||
handler = logging.StreamHandler(stream=sys.stdout)
|
||||
handler.setFormatter(formatter)
|
||||
|
||||
root = logging.getLogger()
|
||||
for existing in root.handlers[:]:
|
||||
root.removeHandler(existing)
|
||||
root.addHandler(handler)
|
||||
root.setLevel(level)
|
||||
|
||||
# Uvicorn sets `propagate = False` on these and attaches its own colourised
|
||||
# handlers, so they must be redirected explicitly or they bypass everything
|
||||
# above. Clearing the handlers and re-enabling propagation routes them
|
||||
# through the root handler like any other logger.
|
||||
for name in ("uvicorn", "uvicorn.error", "uvicorn.access"):
|
||||
uv = logging.getLogger(name)
|
||||
uv.handlers.clear()
|
||||
uv.propagate = True
|
||||
uv.setLevel(level)
|
||||
132
core/rate_limit.py
Normal file
132
core/rate_limit.py
Normal file
@@ -0,0 +1,132 @@
|
||||
"""
|
||||
Rate limiting for the unauthenticated edge.
|
||||
|
||||
**Scope, and why it is narrow.** Every REST/WS/MCP surface is owner-only: an
|
||||
unauthenticated request is rejected by `resolve_bearer`/`is_owner` before any
|
||||
handler runs, and real spend control for outbound calls is
|
||||
`max_concurrent_calls` in `gateway.make_call`. Blanket per-endpoint limits would
|
||||
therefore mostly rate-limit the single legitimate operator. What is genuinely
|
||||
exposed is the handful of `/auth/*` routes that must answer before an identity
|
||||
exists — those are limited here, and nothing else.
|
||||
|
||||
**What this defends against.** Not credential guessing: PATs are
|
||||
`secrets.token_urlsafe(32)` (256 bits) compared by SHA-256 digest, so brute
|
||||
force is not a practical threat. The concern is *unauthenticated work an
|
||||
attacker controls*: `/auth/callback` makes an outbound token-exchange round-trip
|
||||
to Casdoor on every request, and `/auth/me` opens a DB session and runs a token
|
||||
lookup. Both are free to trigger and neither is cheap to serve.
|
||||
|
||||
**Fixed-window, in-process, no dependency.** One operator and a single process
|
||||
mean a shared counter store would be infrastructure without a purpose. The
|
||||
trade-off of a fixed window is a burst of up to 2× the limit across a boundary;
|
||||
that is irrelevant at these thresholds, and it costs one dict lookup with no
|
||||
background task.
|
||||
|
||||
**Client identity is the socket peer, deliberately.** `X-Forwarded-For` is
|
||||
attacker-controlled unless a trusted proxy overwrites it, and this app does not
|
||||
currently establish that trust (`_public_base_url` reads forwarded headers, but
|
||||
only to build URLs). Keying on a spoofable header would let one client present
|
||||
as thousands and make the limiter worse than useless. Behind the estate's
|
||||
reverse proxy this means the limit applies per-proxy rather than per-caller —
|
||||
correct for exhaustion, and honest about what it can enforce. If per-caller
|
||||
limits are ever needed, that needs an explicit trusted-proxy config, not a
|
||||
silent `X-Forwarded-For` read.
|
||||
"""
|
||||
|
||||
import time
|
||||
|
||||
from fastapi import HTTPException, Request
|
||||
|
||||
# Requests allowed per window, per client, per route. Sized to be invisible to a
|
||||
# human — a dashboard load touches /auth/me once — while capping automated
|
||||
# hammering.
|
||||
DEFAULT_LIMIT = 30
|
||||
DEFAULT_WINDOW_SECONDS = 60
|
||||
|
||||
# Stop the bucket store growing without bound under a spray of source addresses.
|
||||
# Eviction is oldest-first and only runs when the cap is exceeded.
|
||||
MAX_TRACKED_CLIENTS = 10_000
|
||||
|
||||
|
||||
class RateLimiter:
|
||||
"""Fixed-window request counter, keyed by (route, client)."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
limit: int = DEFAULT_LIMIT,
|
||||
window_seconds: int = DEFAULT_WINDOW_SECONDS,
|
||||
max_clients: int = MAX_TRACKED_CLIENTS,
|
||||
):
|
||||
self.limit = limit
|
||||
self.window_seconds = window_seconds
|
||||
self.max_clients = max_clients
|
||||
# key -> [window_start, count]. Insertion-ordered, which is what makes
|
||||
# oldest-first eviction a cheap `next(iter(...))`.
|
||||
self._buckets: dict[str, list[float]] = {}
|
||||
|
||||
def check(
|
||||
self, key: str, limit: int | None = None, now: float | None = None
|
||||
) -> tuple[bool, int]:
|
||||
"""Record a hit. Returns (allowed, retry_after_seconds).
|
||||
|
||||
`retry_after` is 0 when allowed, and the seconds remaining in the
|
||||
current window when not. `limit` overrides the instance default for
|
||||
routes that want a tighter cap.
|
||||
"""
|
||||
now = time.monotonic() if now is None else now
|
||||
effective = self.limit if limit is None else limit
|
||||
bucket = self._buckets.get(key)
|
||||
|
||||
if bucket is None or now - bucket[0] >= self.window_seconds:
|
||||
self._buckets.pop(key, None) # re-insert so ordering tracks recency
|
||||
self._buckets[key] = [now, 1]
|
||||
self._evict_if_needed()
|
||||
return True, 0
|
||||
|
||||
bucket[1] += 1
|
||||
if bucket[1] > effective:
|
||||
remaining = self.window_seconds - (now - bucket[0])
|
||||
return False, max(1, int(remaining) + 1)
|
||||
return True, 0
|
||||
|
||||
def _evict_if_needed(self) -> None:
|
||||
while len(self._buckets) > self.max_clients:
|
||||
self._buckets.pop(next(iter(self._buckets)))
|
||||
|
||||
def reset(self) -> None:
|
||||
self._buckets.clear()
|
||||
|
||||
|
||||
def client_key(request: Request, scope: str) -> str:
|
||||
"""Identify the caller for limiting purposes.
|
||||
|
||||
Uses the socket peer, never a forwarded header — see the module docstring.
|
||||
"""
|
||||
client = request.client.host if request.client else "unknown"
|
||||
return f"{scope}:{client}"
|
||||
|
||||
|
||||
_limiter = RateLimiter()
|
||||
|
||||
|
||||
def get_limiter() -> RateLimiter:
|
||||
return _limiter
|
||||
|
||||
|
||||
def rate_limit(scope: str, limit: int | None = None):
|
||||
"""FastAPI dependency factory limiting one route.
|
||||
|
||||
Applied per-route rather than as middleware so the authenticated surfaces —
|
||||
which are already owner-gated — pay nothing.
|
||||
"""
|
||||
|
||||
async def _dependency(request: Request) -> None:
|
||||
allowed, retry_after = get_limiter().check(client_key(request, scope), limit=limit)
|
||||
if not allowed:
|
||||
raise HTTPException(
|
||||
status_code=429,
|
||||
detail="Too many requests",
|
||||
headers={"Retry-After": str(retry_after)},
|
||||
)
|
||||
|
||||
return _dependency
|
||||
@@ -8,6 +8,9 @@ Comprehensive documentation for the Hold Slayer AI telephony gateway.
|
||||
|----------|-------------|
|
||||
| [Architecture](architecture.md) | System architecture, component diagram, data flow |
|
||||
| [Core Engine](core-engine.md) | SIP engine, media pipeline, call manager, event bus |
|
||||
| [Dial Plan](dial-plan.md) | Number normalisation and the emergency-number guard |
|
||||
| [PJSUA2 Build](pjsua2-build.md) | Building the `pjsua2` bindings (not pip-installable) |
|
||||
| [Asterisk Lab](../tests/lab/README.md) | The fake PSTN used for media validation |
|
||||
| [Hold Slayer Service](hold-slayer-service.md) | IVR navigation, hold detection, human detection, transfer |
|
||||
| [Audio Classifier](audio-classifier.md) | Waveform analysis, feature extraction, classification logic |
|
||||
| [Services](services.md) | LLM client, transcription, recording, analytics, notifications |
|
||||
|
||||
@@ -274,10 +274,21 @@ All errors follow a consistent format:
|
||||
| Status Code | Meaning |
|
||||
|-------------|---------|
|
||||
| `400` | Bad request (invalid parameters) |
|
||||
| `401` | Not authenticated (missing or invalid bearer token) |
|
||||
| `403` | Authenticated but not the owner |
|
||||
| `404` | Resource not found (call, flow, device) |
|
||||
| `409` | Conflict (call already ended, device already registered) |
|
||||
| `429` | Rate limited — `/auth/*` routes only. Carries `Retry-After` (seconds) |
|
||||
| `500` | Internal server error |
|
||||
|
||||
`429` applies solely to the unauthenticated `/auth/*` edge: those routes must
|
||||
answer before an identity exists, and each does real work (`/auth/callback`
|
||||
makes an outbound token exchange with Casdoor, `/auth/me` opens a DB session).
|
||||
The owner-gated API is not rate limited — it is already restricted to a single
|
||||
operator. Limits are per client, per route, in a fixed 60-second window; the
|
||||
client is the **socket peer**, so behind a reverse proxy the limit applies
|
||||
per-proxy. See [core/rate_limit.py](../core/rate_limit.py).
|
||||
|
||||
## WebSocket
|
||||
|
||||
### Event Stream
|
||||
|
||||
@@ -4,11 +4,12 @@ Hold Slayer is a single-process async Python application built on FastAPI. It
|
||||
acts as an intelligent B2BUA (Back-to-Back User Agent) sitting between your SIP
|
||||
trunk (PSTN access) and your desk phone/softphone.
|
||||
|
||||
> **Media plane in transition.** The gateway currently signals with Sippy and
|
||||
> intends PJSUA2 to carry media, but PJSUA2 will not surface an RTP stream for
|
||||
> a dialog it does not own — so no audio ever reaches the classifier. The fix
|
||||
> moves *call placement* into PJSUA2 while Sippy keeps the SBC roles. See
|
||||
> [Media plane: why PJSUA2 places the call](#media-plane-why-pjsua2-places-the-call)
|
||||
> **Two SIP engines, selected by `SIP_ENGINE`.** `sippy` (the default) signals
|
||||
> only — PJSUA2 will not surface an RTP stream for a dialog it does not own, so
|
||||
> **no audio reaches the classifier** on that path. `pjsua2`
|
||||
> (`core/pjsua_engine.py`) places the call itself and is the only mode where
|
||||
> audio reaches the classifier; it is opt-in while being proven against the lab.
|
||||
> Read [Media plane: why PJSUA2 places the call](#media-plane-why-pjsua2-places-the-call)
|
||||
> before changing anything in `core/`.
|
||||
|
||||
## System Diagram
|
||||
|
||||
@@ -34,14 +34,36 @@ SSO-disabled-off-loopback.
|
||||
| `SIP_TRUNK_DID` | Your phone number (E.164) | — | Yes |
|
||||
| `SIP_TRUNK_TRANSPORT` | Transport protocol (`udp`, `tcp`, `tls`) | `udp` | No |
|
||||
|
||||
### Server
|
||||
|
||||
| Variable | Description | Default | Required |
|
||||
|----------|-------------|---------|----------|
|
||||
| `HOST` | Bind address. Off-loopback requires `CASDOOR_ENABLED=true` | `0.0.0.0` | No |
|
||||
| `PORT` | Bind port | `8000` | No |
|
||||
| `DEBUG` | SQLAlchemy echo + uvicorn reload | `false` | No |
|
||||
| `LOG_LEVEL` | Root log level (`debug`/`info`/`warning`/`error`) | `info` | No |
|
||||
| `LOG_FORMAT` | `text` (human-readable) or `json` (structured, for Loki) | `text` | No |
|
||||
|
||||
`LOG_FORMAT=json` renders one JSON object per line, including uvicorn's access
|
||||
log — `method`, `path`, `status_code` (numeric, so it can be range-filtered) and
|
||||
`client_addr` arrive as queryable fields rather than a formatted string. The
|
||||
Docker image sets it; `text` is the default so local development stays readable.
|
||||
|
||||
### Safety
|
||||
|
||||
| Variable | Description | Default | Required |
|
||||
|----------|-------------|---------|----------|
|
||||
| `MAX_CONCURRENT_CALLS` | Cap on simultaneous outbound calls | `4` | No |
|
||||
| `USE_MOCK_SIP` | Run the mock SIP engine — no real calls. Must be asked for explicitly; an unconfigured trunk without it fails startup | `false` | No |
|
||||
| `SIP_ENGINE` | `sippy` (signalling only — no audio reaches the classifier) or `pjsua2` (call control + media) | `sippy` | No |
|
||||
|
||||
### Gateway
|
||||
|
||||
| Variable | Description | Default | Required |
|
||||
|----------|-------------|---------|----------|
|
||||
| `GATEWAY_SIP_PORT` | Port for device SIP registration | `5080` | No |
|
||||
| `GATEWAY_RTP_PORT_MIN` | Minimum RTP port | `10000` | No |
|
||||
| `GATEWAY_RTP_PORT_MAX` | Maximum RTP port | `20000` | No |
|
||||
| `GATEWAY_HOST` | Bind address | `0.0.0.0` | No |
|
||||
| `GATEWAY_SIP_HOST` | Bind address for the device-registration listener | `0.0.0.0` | No |
|
||||
| `GATEWAY_SIP_PORT` | Port for device SIP registration | `5060` | No |
|
||||
| `GATEWAY_SIP_DOMAIN` | SIP domain devices register against | `gateway.local` | No |
|
||||
|
||||
### LLM
|
||||
|
||||
@@ -65,7 +87,7 @@ SSO-disabled-off-loopback.
|
||||
|
||||
| Variable | Description | Default | Required |
|
||||
|----------|-------------|---------|----------|
|
||||
| `DATABASE_URL` | PostgreSQL or SQLite connection string | `sqlite+aiosqlite:///./hold_slayer.db` | No |
|
||||
| `DATABASE_URL` | PostgreSQL connection string. Startup exits with a readable error if unset | — | Yes |
|
||||
|
||||
### Notifications
|
||||
|
||||
@@ -73,6 +95,18 @@ SSO-disabled-off-loopback.
|
||||
|----------|-------------|---------|----------|
|
||||
| `NOTIFY_SMS_NUMBER` | Phone number for SMS alerts (E.164) | — | No |
|
||||
|
||||
### Receptionist
|
||||
|
||||
| Variable | Description | Default | Required |
|
||||
|----------|-------------|---------|----------|
|
||||
| `RECEPTIONIST_ENABLED` | Answer inbound calls with the AI receptionist | `true` | No |
|
||||
| `RECEPTIONIST_GREETING_TEMPLATE` | Spoken greeting | `"Hi, you've reached Robert's line. Who's calling, and what's this about?"` | No |
|
||||
| `RECEPTIONIST_MESSAGE_PROMPT` | Spoken prompt before recording a message | `"Please leave your message after the tone."` | No |
|
||||
| `RECEPTIONIST_LLM_PERSONA` | System prompt shaping the receptionist's decisions | See `config.py` | No |
|
||||
| `RECEPTIONIST_LISTEN_TIMEOUT_S` | Seconds to wait for the caller to speak | `15.0` | No |
|
||||
| `RECEPTIONIST_END_OF_UTTERANCE_SILENCE_S` | Silence marking the end of a turn | `1.2` | No |
|
||||
| `RECEPTIONIST_MESSAGE_MAX_SECONDS` | Voicemail cap | `90` | No |
|
||||
|
||||
### Audio Classifier
|
||||
|
||||
| Variable | Description | Default | Required |
|
||||
|
||||
24
main.py
24
main.py
@@ -26,6 +26,7 @@ from api import call_flows, call_history, calls, devices, routing, tokens, webso
|
||||
from auth import get_current_owner, init_jwks_client, is_owner, resolve_from_header_or_query
|
||||
from config import Settings, get_settings
|
||||
from core.gateway import AIPSTNGateway, build_sip_engine
|
||||
from core.logging_config import configure_logging
|
||||
from db.database import close_db, init_db, session_scope
|
||||
from mcp_server.server import create_mcp_server
|
||||
from models.call import CallMode
|
||||
@@ -39,13 +40,12 @@ from services.routing import RoutingService
|
||||
from services.transcription import TranscriptionService
|
||||
from services.tts import TTSService
|
||||
|
||||
# Configure logging
|
||||
logging.basicConfig(
|
||||
level=logging.INFO,
|
||||
format="%(asctime)s | %(levelname)-7s | %(name)s | %(message)s",
|
||||
datefmt="%H:%M:%S",
|
||||
stream=sys.stdout,
|
||||
)
|
||||
# Configure logging at import so anything logged during module import and
|
||||
# startup config checks is formatted. Uvicorn installs its own handlers *after*
|
||||
# importing this module, so `lifespan` calls `configure_logging` again to take
|
||||
# them over — see core/logging_config.py.
|
||||
_startup_settings = get_settings()
|
||||
configure_logging(_startup_settings.log_format, _startup_settings.log_level)
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
@@ -150,6 +150,12 @@ def _check_startup_config(settings: Settings) -> None:
|
||||
async def lifespan(app: FastAPI):
|
||||
"""Startup: Initialize database, SIP engine, and services."""
|
||||
settings = get_settings()
|
||||
|
||||
# Re-apply: under `uvicorn main:app` the server installs its own handlers on
|
||||
# `uvicorn`/`uvicorn.access` after importing this module, which would emit
|
||||
# colourised text alongside our JSON. This takes them back over.
|
||||
configure_logging(settings.log_format, settings.log_level)
|
||||
|
||||
_check_startup_config(settings)
|
||||
|
||||
# Prefetch Casdoor's JWKS so the first authenticated request doesn't pay
|
||||
@@ -622,4 +628,8 @@ if __name__ == "__main__":
|
||||
port=settings.port,
|
||||
reload=settings.debug,
|
||||
log_level=settings.log_level,
|
||||
# Suppress uvicorn's own dictConfig: ours is installed at import and in
|
||||
# lifespan, and letting uvicorn apply its default would attach a second,
|
||||
# colourised handler — every line twice, one of them not JSON.
|
||||
log_config=None,
|
||||
)
|
||||
|
||||
@@ -1,24 +1,17 @@
|
||||
; Minimal Asterisk core config for the lab.
|
||||
[directories](!)
|
||||
astetcdir => /etc/asterisk
|
||||
astmoddir => /usr/lib/asterisk/modules
|
||||
astvarlibdir => /var/lib/asterisk
|
||||
astdbdir => /var/lib/asterisk
|
||||
astkeydir => /var/lib/asterisk
|
||||
astdatadir => /var/lib/asterisk
|
||||
astagidir => /var/lib/asterisk/agi-bin
|
||||
astspooldir => /var/spool/asterisk
|
||||
astrundir => /var/run/asterisk
|
||||
astlogdir => /var/log/asterisk
|
||||
astsbindir => /usr/sbin
|
||||
|
||||
;
|
||||
; Deliberately does NOT set [directories] or runuser/rungroup: the image's
|
||||
; compiled-in defaults are correct, and it runs as the `asterisk` user via a
|
||||
; USER directive. Overriding either risks breaking the container for no gain
|
||||
; (an earlier version of this file did both).
|
||||
[options]
|
||||
; Log to stdout so Docker's json-file driver captures it and Alloy ships it.
|
||||
; A file-based log inside the container would be invisible to Loki.
|
||||
; Log to stdout so Docker's json-file driver captures it and Alloy ships it
|
||||
; to Loki. A file-based log inside the container would be invisible.
|
||||
verbose = 3
|
||||
debug = 0
|
||||
; No ANSI colour. Asterisk colourises the console by default and the escape
|
||||
; codes travel through Docker into Loki, where every line arrives wrapped in
|
||||
; \x1b[0;30m — unreadable in Grafana and awkward to filter on. This setting
|
||||
; lives here, not in logger.conf, and only takes effect if this file is
|
||||
; actually mounted into the container.
|
||||
nocolor = yes
|
||||
dumpcore = no
|
||||
; Never run as root inside the container.
|
||||
runuser = asterisk
|
||||
rungroup = asterisk
|
||||
|
||||
@@ -3,6 +3,22 @@
|
||||
; the container would put the logs where nothing can see them.
|
||||
[general]
|
||||
dateformat = %F %T
|
||||
; Colour is disabled in asterisk.conf (`nocolor = yes`), not here — Asterisk
|
||||
; colourises the console by default and the escape codes travel through Docker
|
||||
; into Loki, where every line arrives wrapped in \x1b[0;30m. That file must be
|
||||
; mounted for the setting to take effect.
|
||||
|
||||
[logfiles]
|
||||
console => notice,warning,error
|
||||
; Warnings and errors only. That covers what matters when a lab call
|
||||
; misbehaves: failed authentication, no-matching-endpoint, playback failures.
|
||||
;
|
||||
; `notice` and `verbose` are both excluded because the "Remote UNIX
|
||||
; connection" pairs the healthcheck generates arrive on those channels, and
|
||||
; they swamped everything else (the Loki stream measured 100% healthcheck
|
||||
; noise before this). The healthcheck itself is also reduced to one CLI call
|
||||
; on a 60s interval in docker-compose — the two changes work together.
|
||||
;
|
||||
; To trace a call's dialplan execution, raise verbosity at runtime rather than
|
||||
; leaving it on:
|
||||
; asterisk -rx "core set verbose 3"
|
||||
console => warning,error
|
||||
|
||||
@@ -21,8 +21,22 @@ services:
|
||||
- ./dialplan/pjsip.local.conf:/etc/asterisk/pjsip.conf:ro
|
||||
- ./dialplan/rtp.local.conf:/etc/asterisk/rtp.conf:ro
|
||||
- ./dialplan/logger.conf:/etc/asterisk/logger.conf:ro
|
||||
# Carries `nocolor = yes`: without it every log line reaches Loki
|
||||
# wrapped in ANSI escape codes.
|
||||
- ./dialplan/asterisk.conf:/etc/asterisk/asterisk.conf:ro
|
||||
# The image ships no sound files at all. These are generated by
|
||||
# sounds/generate.py; Asterisk resolves Playback(lab-music) to
|
||||
# lab-music.sln here (8kHz signed-linear, no transcoding).
|
||||
- ./sounds:/var/lib/asterisk/sounds/en:ro
|
||||
# The image's default command is `-vvvdddf` — verbosity 3 and debug 3
|
||||
# forced on the command line, which overrides both asterisk.conf and
|
||||
# logger.conf. That makes every healthcheck CLI connection log a
|
||||
# "Remote UNIX connection" pair: ~2900 lines/day of pure noise that
|
||||
# completely buried the real SIP events in Loki.
|
||||
#
|
||||
# -f foreground (required: Docker needs PID 1 to stay), -T timestamps,
|
||||
# -W colour off, -U run as asterisk, -p realtime priority. No -v, no -d:
|
||||
# warnings and errors still log, and verbosity can be raised at runtime
|
||||
# with `asterisk -rx "core set verbose 3"` when tracing a call.
|
||||
command: ["/usr/sbin/asterisk", "-f", "-T", "-W", "-U", "asterisk", "-p"]
|
||||
restart: unless-stopped
|
||||
|
||||
187
tests/test_graceful_degradation.py
Normal file
187
tests/test_graceful_degradation.py
Normal file
@@ -0,0 +1,187 @@
|
||||
"""
|
||||
Graceful-degradation tests.
|
||||
|
||||
Every external dependency — STT, LLM, TTS — is reachable over the network and
|
||||
can be down. The gateway's rule is that a dead dependency degrades the call
|
||||
rather than killing it, and says so: each failure publishes an `ERROR` event
|
||||
naming the service, so a down Speaches reads as "transcription failed" rather
|
||||
than "the AI is making bad decisions".
|
||||
|
||||
The behaviour is already implemented across the services; these tests exist so a
|
||||
later refactor can't quietly remove it. The failure mode being guarded against is
|
||||
silent: an un-caught exception in one of these paths aborts a live phone call.
|
||||
"""
|
||||
|
||||
from unittest.mock import AsyncMock, MagicMock
|
||||
|
||||
import pytest
|
||||
|
||||
from config import Settings
|
||||
from core.event_bus import EventBus
|
||||
from models.events import EventType
|
||||
|
||||
|
||||
def _gateway():
|
||||
"""A gateway stand-in with a real event bus, so events can be asserted."""
|
||||
gw = MagicMock()
|
||||
gw.settings = Settings(database_url="sqlite+aiosqlite:///:memory:")
|
||||
gw.event_bus = EventBus()
|
||||
gw.call_manager = MagicMock()
|
||||
gw.call_manager.add_transcript = AsyncMock()
|
||||
return gw
|
||||
|
||||
|
||||
def _hold_slayer(gateway, transcription):
|
||||
from services.audio_classifier import AudioClassifier
|
||||
from services.hold_slayer import HoldSlayerService
|
||||
|
||||
return HoldSlayerService(
|
||||
gateway=gateway,
|
||||
call_manager=gateway.call_manager,
|
||||
sip_engine=MagicMock(),
|
||||
classifier=AudioClassifier(gateway.settings.classifier),
|
||||
transcription=transcription,
|
||||
settings=gateway.settings,
|
||||
)
|
||||
|
||||
|
||||
async def _errors_for(bus: EventBus, coro):
|
||||
"""Run `coro` while subscribed, returning the ERROR events it published."""
|
||||
import asyncio
|
||||
|
||||
sub = bus.subscribe(event_types={EventType.ERROR})
|
||||
try:
|
||||
result = await coro
|
||||
seen = []
|
||||
while True:
|
||||
try:
|
||||
seen.append(sub._queue.get_nowait())
|
||||
except asyncio.QueueEmpty:
|
||||
break
|
||||
return result, seen
|
||||
finally:
|
||||
bus.unsubscribe(sub)
|
||||
|
||||
|
||||
class TestClassifierWithoutSTT:
|
||||
"""The classifier is spectral: it must not depend on STT at all."""
|
||||
|
||||
def _classifier(self):
|
||||
from services.audio_classifier import AudioClassifier
|
||||
|
||||
return AudioClassifier(Settings(database_url="sqlite+aiosqlite:///:memory:").classifier)
|
||||
|
||||
def test_classify_takes_only_audio(self):
|
||||
# A transcript parameter would make STT a hard dependency of hold
|
||||
# detection — the thing this checkbox is about.
|
||||
import inspect
|
||||
|
||||
params = set(inspect.signature(self._classifier().classify_chunk).parameters)
|
||||
assert params == {"audio_data"}
|
||||
|
||||
async def test_classifies_with_no_stt_service_anywhere(self):
|
||||
# Silence is the cheapest deterministic input; the point is that a
|
||||
# classification is produced at all with no STT in the picture.
|
||||
result = await self._classifier().classify(b"\x00\x00" * 16000)
|
||||
assert result.audio_type is not None
|
||||
|
||||
|
||||
class TestTranscriptionDegradation:
|
||||
async def test_hold_slayer_transcribe_returns_empty_on_failure(self):
|
||||
gw = _gateway()
|
||||
stt = MagicMock()
|
||||
stt.transcribe = AsyncMock(side_effect=RuntimeError("Connection refused"))
|
||||
svc = _hold_slayer(gw, stt)
|
||||
|
||||
text, errors = await _errors_for(gw.event_bus, svc._transcribe("call-1", b"\x00" * 320))
|
||||
|
||||
assert text == "" # empty transcript, not an exception
|
||||
assert len(errors) == 1
|
||||
assert errors[0].data["service"] == "transcription"
|
||||
|
||||
async def test_error_event_names_the_service(self):
|
||||
# "transcription failed" vs "the AI decided badly" — the whole reason
|
||||
# transcribe() raises instead of swallowing.
|
||||
gw = _gateway()
|
||||
stt = MagicMock()
|
||||
stt.transcribe = AsyncMock(side_effect=RuntimeError("Connection refused"))
|
||||
svc = _hold_slayer(gw, stt)
|
||||
|
||||
_, errors = await _errors_for(gw.event_bus, svc._transcribe("call-1", b"\x00" * 320))
|
||||
assert "Connection refused" in errors[0].data["error"]
|
||||
|
||||
async def test_service_error_survives_a_dead_event_bus(self):
|
||||
# Degradation reporting must not itself become a failure path.
|
||||
gw = _gateway()
|
||||
gw.event_bus.publish = AsyncMock(side_effect=RuntimeError("bus down"))
|
||||
stt = MagicMock()
|
||||
stt.transcribe = AsyncMock(side_effect=RuntimeError("stt down"))
|
||||
svc = _hold_slayer(gw, stt)
|
||||
|
||||
assert await svc._transcribe("call-1", b"\x00" * 320) == ""
|
||||
|
||||
async def test_transcription_marks_itself_unavailable(self):
|
||||
# /health reads this flag; a failure that doesn't record itself makes
|
||||
# the probe lie.
|
||||
import httpx
|
||||
|
||||
from services.transcription import TranscriptionService
|
||||
|
||||
svc = TranscriptionService(Settings(database_url="sqlite+aiosqlite:///:memory:").speaches)
|
||||
client = MagicMock()
|
||||
client.post = AsyncMock(side_effect=httpx.ConnectError("refused"))
|
||||
svc._client = client
|
||||
svc._client.is_closed = False
|
||||
|
||||
with pytest.raises(Exception):
|
||||
await svc.transcribe(b"\x00" * 320)
|
||||
assert svc.available is False
|
||||
|
||||
|
||||
class TestReceptionistDegradation:
|
||||
def _receptionist(self, gateway, **kw):
|
||||
from services.receptionist import ReceptionistService
|
||||
|
||||
return ReceptionistService(gateway=gateway, **kw)
|
||||
|
||||
async def test_llm_failure_falls_back_to_a_usable_decision(self):
|
||||
gw = _gateway()
|
||||
svc = self._receptionist(gw)
|
||||
call = MagicMock(id="call-1", remote_number="+15551234567")
|
||||
|
||||
llm = MagicMock()
|
||||
llm.chat_json = AsyncMock(side_effect=RuntimeError("LLM down"))
|
||||
import services.llm_client as llm_mod
|
||||
|
||||
original = llm_mod.get_llm
|
||||
llm_mod.get_llm = lambda: llm
|
||||
try:
|
||||
result, errors = await _errors_for(
|
||||
gw.event_bus, svc._classify(call, "I need to speak to someone", None)
|
||||
)
|
||||
finally:
|
||||
llm_mod.get_llm = original
|
||||
|
||||
# A decision still comes back, so the call can proceed.
|
||||
assert result["recommended_action"] in {"ring", "message", "reject"}
|
||||
assert errors[0].data["service"] == "llm"
|
||||
|
||||
async def test_no_transcription_service_yields_empty_not_crash(self):
|
||||
# transcription=None is a valid wiring (STT not configured).
|
||||
gw = _gateway()
|
||||
svc = self._receptionist(gw, transcription=None)
|
||||
assert svc.transcription is None
|
||||
|
||||
|
||||
class TestHealthReportsDegradation:
|
||||
"""A degraded gateway must read as degraded — /health may not lie."""
|
||||
|
||||
def test_availability_helper_distinguishes_unknown_from_down(self):
|
||||
# Four distinct states, because "not wired up" and "wired up but the
|
||||
# remote is refusing connections" are different operator problems.
|
||||
import main
|
||||
|
||||
assert main._availability(None) == "not attached"
|
||||
assert main._availability(MagicMock(available=None)) == "unknown (no requests yet)"
|
||||
assert main._availability(MagicMock(available=True)) == "ok"
|
||||
assert main._availability(MagicMock(available=False)) == "unreachable"
|
||||
183
tests/test_logging_config.py
Normal file
183
tests/test_logging_config.py
Normal file
@@ -0,0 +1,183 @@
|
||||
"""
|
||||
Structured-logging tests.
|
||||
|
||||
The two things worth guarding are the ones that are easy to break silently:
|
||||
uvicorn's access logger must actually route through our formatter (it sets
|
||||
`propagate = False` and brings its own handler), and its arg tuple must land as
|
||||
real JSON fields rather than a pre-formatted string.
|
||||
"""
|
||||
|
||||
import json
|
||||
import logging
|
||||
import logging.config
|
||||
|
||||
import pytest
|
||||
|
||||
from config import Settings
|
||||
from core.logging_config import JSONFormatter, configure_logging
|
||||
|
||||
|
||||
def _record(name="test.logger", level=logging.INFO, msg="hello", args=None, **extra):
|
||||
record = logging.LogRecord(
|
||||
name=name, level=level, pathname=__file__, lineno=1, msg=msg, args=args, exc_info=None
|
||||
)
|
||||
for key, value in extra.items():
|
||||
setattr(record, key, value)
|
||||
return record
|
||||
|
||||
|
||||
def _emit(record):
|
||||
return json.loads(JSONFormatter().format(record))
|
||||
|
||||
|
||||
class TestJSONFormatter:
|
||||
def test_emits_single_line_json_with_core_fields(self):
|
||||
out = JSONFormatter().format(_record())
|
||||
assert "\n" not in out
|
||||
payload = json.loads(out)
|
||||
assert payload["msg"] == "hello"
|
||||
assert payload["level"] == "INFO"
|
||||
assert payload["logger"] == "test.logger"
|
||||
|
||||
def test_timestamp_is_utc_rfc3339_with_date(self):
|
||||
# logging's default asctime is local-time and date-less, which is
|
||||
# exactly what makes text logs hard to correlate in Loki.
|
||||
ts = _emit(_record())["ts"]
|
||||
assert ts.endswith("+00:00")
|
||||
assert "T" in ts
|
||||
|
||||
def test_interpolates_message_args(self):
|
||||
assert _emit(_record(msg="call %s ended", args=("abc123",)))["msg"] == "call abc123 ended"
|
||||
|
||||
def test_extra_fields_are_promoted(self):
|
||||
payload = _emit(_record(call_id="c-1", duration=12.5))
|
||||
assert payload["call_id"] == "c-1"
|
||||
assert payload["duration"] == 12.5
|
||||
|
||||
def test_uvicorn_color_message_is_dropped(self):
|
||||
# Uvicorn ships an ANSI-coloured copy of the message via extra=; letting
|
||||
# it through puts escape codes in Loki.
|
||||
payload = _emit(
|
||||
_record(name="uvicorn.error", msg="Started", color_message="\x1b[36mStarted\x1b[0m")
|
||||
)
|
||||
assert "color_message" not in payload
|
||||
assert "\x1b" not in json.dumps(payload)
|
||||
|
||||
def test_non_json_native_extra_is_stringified(self):
|
||||
payload = _emit(_record(obj=object()))
|
||||
assert isinstance(payload["obj"], str)
|
||||
|
||||
def test_secretstr_extra_stays_masked(self):
|
||||
# Config secrets are SecretStr precisely so an accidental log can't leak
|
||||
# them; stringification must preserve that.
|
||||
from pydantic import SecretStr
|
||||
|
||||
payload = _emit(_record(secret=SecretStr("hs_pat_supersecret")))
|
||||
assert "supersecret" not in json.dumps(payload)
|
||||
|
||||
def test_exception_is_captured(self):
|
||||
try:
|
||||
raise ValueError("boom")
|
||||
except ValueError:
|
||||
import sys
|
||||
|
||||
record = _record(level=logging.ERROR, msg="failed")
|
||||
record.exc_info = sys.exc_info()
|
||||
payload = _emit(record)
|
||||
assert "ValueError: boom" in payload["exc"]
|
||||
|
||||
def test_thread_name_included_only_off_main_thread(self):
|
||||
# Which execution context logged a line is the first question when
|
||||
# debugging a call across the asyncio/Sippy/PJSUA2 boundary.
|
||||
on_main = _record()
|
||||
on_main.threadName = "MainThread"
|
||||
assert "thread" not in _emit(on_main)
|
||||
|
||||
off_main = _record()
|
||||
off_main.threadName = "sippy-ed"
|
||||
assert _emit(off_main)["thread"] == "sippy-ed"
|
||||
|
||||
|
||||
class TestAccessLogFields:
|
||||
def _access(self, status=200):
|
||||
return _emit(
|
||||
_record(
|
||||
name="uvicorn.access",
|
||||
msg='%s - "%s %s HTTP/%s" %d',
|
||||
args=("127.0.0.1:5050", "GET", "/api/v1/calls", "1.1", status),
|
||||
)
|
||||
)
|
||||
|
||||
def test_arg_tuple_becomes_structured_fields(self):
|
||||
payload = self._access()
|
||||
assert payload["method"] == "GET"
|
||||
assert payload["path"] == "/api/v1/calls"
|
||||
assert payload["client_addr"] == "127.0.0.1:5050"
|
||||
assert payload["http_version"] == "1.1"
|
||||
|
||||
def test_status_code_is_a_number_not_a_string(self):
|
||||
# So Loki can range-filter on it (status_code >= 500).
|
||||
assert self._access(503)["status_code"] == 503
|
||||
|
||||
def test_access_line_is_not_pre_formatted(self):
|
||||
# The whole point: no interpolated request line to regex back apart.
|
||||
assert "msg" not in self._access()
|
||||
|
||||
def test_unexpected_arg_shape_falls_back_to_message(self):
|
||||
# A logging path that raises would take out the request; degrade instead.
|
||||
payload = _emit(_record(name="uvicorn.access", msg="just a string", args=None))
|
||||
assert payload["msg"] == "just a string"
|
||||
|
||||
|
||||
class TestConfigureLogging:
|
||||
@pytest.fixture(autouse=True)
|
||||
def _restore(self):
|
||||
root = logging.getLogger()
|
||||
saved = root.handlers[:], root.level
|
||||
yield
|
||||
root.handlers[:] = saved[0]
|
||||
root.setLevel(saved[1])
|
||||
|
||||
def _uvicorn_defaults(self):
|
||||
"""Reproduce what uvicorn does to its loggers at startup."""
|
||||
from uvicorn.config import LOGGING_CONFIG
|
||||
|
||||
logging.config.dictConfig(LOGGING_CONFIG)
|
||||
|
||||
def test_takes_over_uvicorn_handlers(self):
|
||||
self._uvicorn_defaults()
|
||||
access = logging.getLogger("uvicorn.access")
|
||||
assert access.propagate is False # uvicorn's default, the problem
|
||||
|
||||
configure_logging("json", "info")
|
||||
assert access.propagate is True
|
||||
assert access.handlers == []
|
||||
|
||||
def test_json_mode_installs_json_formatter(self):
|
||||
configure_logging("json", "info")
|
||||
assert isinstance(logging.getLogger().handlers[0].formatter, JSONFormatter)
|
||||
|
||||
def test_text_mode_does_not(self):
|
||||
configure_logging("text", "info")
|
||||
assert not isinstance(logging.getLogger().handlers[0].formatter, JSONFormatter)
|
||||
|
||||
def test_is_idempotent(self):
|
||||
# Called at import and again in lifespan; a second call must replace the
|
||||
# handler, not add one, or every line is emitted twice.
|
||||
configure_logging("json", "info")
|
||||
configure_logging("json", "info")
|
||||
assert len(logging.getLogger().handlers) == 1
|
||||
|
||||
def test_respects_log_level(self):
|
||||
configure_logging("json", "warning")
|
||||
assert logging.getLogger().level == logging.WARNING
|
||||
|
||||
|
||||
class TestSettings:
|
||||
def test_defaults_to_text(self):
|
||||
# A JSON-only default would make local development worse.
|
||||
assert Settings(database_url="sqlite+aiosqlite:///:memory:").log_format == "text"
|
||||
|
||||
def test_reads_log_format_env(self, monkeypatch):
|
||||
monkeypatch.setenv("LOG_FORMAT", "json")
|
||||
assert Settings(database_url="sqlite+aiosqlite:///:memory:").log_format == "json"
|
||||
185
tests/test_rate_limit.py
Normal file
185
tests/test_rate_limit.py
Normal file
@@ -0,0 +1,185 @@
|
||||
"""
|
||||
Rate-limiting tests.
|
||||
|
||||
The limiter guards the unauthenticated `/auth/*` edge — the only routes that
|
||||
must answer before an identity exists. Everything else is owner-gated, so a
|
||||
limit there would mostly throttle the single legitimate operator.
|
||||
|
||||
The properties worth pinning: the cap actually blocks, windows expire, clients
|
||||
and routes don't share a bucket, the store can't grow without bound, and
|
||||
identity comes from the socket peer rather than a spoofable header.
|
||||
"""
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
from fastapi import Depends, FastAPI
|
||||
|
||||
import main
|
||||
from core.rate_limit import RateLimiter, client_key, get_limiter, rate_limit
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _clean_limiter():
|
||||
get_limiter().reset()
|
||||
yield
|
||||
get_limiter().reset()
|
||||
|
||||
|
||||
class TestRateLimiter:
|
||||
def test_allows_up_to_the_limit(self):
|
||||
rl = RateLimiter(limit=3, window_seconds=60)
|
||||
assert [rl.check("k", now=100.0)[0] for _ in range(3)] == [True, True, True]
|
||||
|
||||
def test_blocks_past_the_limit(self):
|
||||
rl = RateLimiter(limit=3, window_seconds=60)
|
||||
for _ in range(3):
|
||||
rl.check("k", now=100.0)
|
||||
allowed, retry_after = rl.check("k", now=100.0)
|
||||
assert allowed is False
|
||||
assert retry_after > 0
|
||||
|
||||
def test_window_expiry_resets_the_count(self):
|
||||
rl = RateLimiter(limit=2, window_seconds=60)
|
||||
rl.check("k", now=100.0)
|
||||
rl.check("k", now=100.0)
|
||||
assert rl.check("k", now=100.0)[0] is False
|
||||
# A full window later, the caller is welcome again.
|
||||
assert rl.check("k", now=161.0)[0] is True
|
||||
|
||||
def test_retry_after_shrinks_as_the_window_drains(self):
|
||||
rl = RateLimiter(limit=1, window_seconds=60)
|
||||
rl.check("k", now=100.0)
|
||||
early = rl.check("k", now=110.0)[1]
|
||||
late = rl.check("k", now=150.0)[1]
|
||||
assert early > late >= 1
|
||||
|
||||
def test_clients_do_not_share_a_bucket(self):
|
||||
rl = RateLimiter(limit=1, window_seconds=60)
|
||||
assert rl.check("auth:me:1.1.1.1", now=100.0)[0] is True
|
||||
# A different client must be unaffected by the first one's usage.
|
||||
assert rl.check("auth:me:2.2.2.2", now=100.0)[0] is True
|
||||
|
||||
def test_routes_do_not_share_a_bucket(self):
|
||||
rl = RateLimiter(limit=1, window_seconds=60)
|
||||
assert rl.check("auth:me:1.1.1.1", now=100.0)[0] is True
|
||||
assert rl.check("auth:callback:1.1.1.1", now=100.0)[0] is True
|
||||
|
||||
def test_per_call_limit_overrides_the_default(self):
|
||||
rl = RateLimiter(limit=100, window_seconds=60)
|
||||
rl.check("k", limit=1, now=100.0)
|
||||
assert rl.check("k", limit=1, now=100.0)[0] is False
|
||||
|
||||
def test_bucket_store_is_bounded(self):
|
||||
# Otherwise a spray of source addresses is itself a memory exhaustion
|
||||
# vector — the thing the limiter exists to prevent.
|
||||
rl = RateLimiter(limit=5, window_seconds=60, max_clients=10)
|
||||
for i in range(50):
|
||||
rl.check(f"client-{i}", now=100.0 + i)
|
||||
assert len(rl._buckets) <= 10
|
||||
|
||||
def test_eviction_drops_oldest_first(self):
|
||||
rl = RateLimiter(limit=5, window_seconds=600, max_clients=3)
|
||||
for i in range(4):
|
||||
rl.check(f"client-{i}", now=100.0 + i)
|
||||
assert "client-0" not in rl._buckets
|
||||
assert "client-3" in rl._buckets
|
||||
|
||||
|
||||
class TestClientKey:
|
||||
def _request(self, peer: str | None, headers: dict | None = None):
|
||||
scope = {
|
||||
"type": "http",
|
||||
"method": "GET",
|
||||
"path": "/auth/me",
|
||||
"headers": [(k.lower().encode(), v.encode()) for k, v in (headers or {}).items()],
|
||||
"client": (peer, 12345) if peer else None,
|
||||
}
|
||||
from starlette.requests import Request
|
||||
|
||||
return Request(scope)
|
||||
|
||||
def test_uses_the_socket_peer(self):
|
||||
assert client_key(self._request("10.0.0.5"), "auth:me") == "auth:me:10.0.0.5"
|
||||
|
||||
def test_ignores_x_forwarded_for(self):
|
||||
# Trusting a spoofable header would let one client present as
|
||||
# thousands, making the limiter worse than useless.
|
||||
key = client_key(
|
||||
self._request("10.0.0.5", {"X-Forwarded-For": "1.2.3.4"}), "auth:me"
|
||||
)
|
||||
assert key == "auth:me:10.0.0.5"
|
||||
assert "1.2.3.4" not in key
|
||||
|
||||
def test_missing_peer_does_not_crash(self):
|
||||
assert client_key(self._request(None), "auth:me") == "auth:me:unknown"
|
||||
|
||||
|
||||
class TestDependency:
|
||||
"""The FastAPI integration: a 429 with a Retry-After header."""
|
||||
|
||||
def _app(self, limit=2):
|
||||
app = FastAPI()
|
||||
|
||||
@app.get("/limited", dependencies=[Depends(rate_limit("test", limit=limit))])
|
||||
async def limited():
|
||||
return {"ok": True}
|
||||
|
||||
@app.get("/unlimited")
|
||||
async def unlimited():
|
||||
return {"ok": True}
|
||||
|
||||
return app
|
||||
|
||||
async def _get(self, app, path, n=1):
|
||||
transport = httpx.ASGITransport(app=app)
|
||||
async with httpx.AsyncClient(transport=transport, base_url="http://test") as c:
|
||||
return [await c.get(path) for _ in range(n)]
|
||||
|
||||
async def test_returns_429_past_the_limit(self):
|
||||
responses = await self._get(self._app(limit=2), "/limited", n=3)
|
||||
assert [r.status_code for r in responses] == [200, 200, 429]
|
||||
|
||||
async def test_429_carries_retry_after(self):
|
||||
responses = await self._get(self._app(limit=1), "/limited", n=2)
|
||||
blocked = responses[-1]
|
||||
assert blocked.status_code == 429
|
||||
assert int(blocked.headers["retry-after"]) >= 1
|
||||
|
||||
async def test_unlimited_routes_are_untouched(self):
|
||||
responses = await self._get(self._app(limit=1), "/unlimited", n=10)
|
||||
assert {r.status_code for r in responses} == {200}
|
||||
|
||||
|
||||
class TestAuthRoutesAreLimited:
|
||||
"""The wiring: the unauthenticated edge is covered, the rest is not."""
|
||||
|
||||
def _is_limited(self, path: str) -> bool:
|
||||
"""True if the route carries a dependency built by `rate_limit`.
|
||||
|
||||
Identified by the closure's qualname rather than a string search, so
|
||||
this can't pass on an unrelated dependency that happens to stringify
|
||||
similarly.
|
||||
"""
|
||||
for route in main.app.routes:
|
||||
if getattr(route, "path", None) != path:
|
||||
continue
|
||||
return any(
|
||||
getattr(d.dependency, "__qualname__", "").startswith("rate_limit")
|
||||
for d in getattr(route, "dependencies", [])
|
||||
)
|
||||
raise AssertionError(f"route {path} not found")
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"path", ["/auth/login", "/auth/callback", "/auth/me", "/auth/refresh-callback"]
|
||||
)
|
||||
def test_unauthenticated_auth_routes_are_limited(self, path):
|
||||
assert self._is_limited(path)
|
||||
|
||||
def test_owner_gated_routes_are_not_limited(self):
|
||||
# They're already behind is_owner; limiting them would throttle the
|
||||
# single legitimate operator.
|
||||
assert not self._is_limited("/api/v1/calls/active")
|
||||
|
||||
def test_logout_is_not_limited(self):
|
||||
# Pure redirect builder — no I/O, nothing to exhaust.
|
||||
assert not self._is_limited("/auth/logout")
|
||||
Reference in New Issue
Block a user