Hermes Wiki
LocalzDocs/Minimal-to-Production-Security-Implementation

Minimal to Production Path — Security Implementation

Reconciliation note (2026-08-28): this doc predates the stack decisions finalized in Stack-Finalization-Draft.md. Two things below are now stale and have been updated to match: the datastore (this doc originally specified MongoDB for messages/threads — the decided production datastore is Postgres-only, JSONB for flexible-schema content like message bodies/metadata, per Stack-Finalization-Draft's "Mongo vs Postgres" decision) and the "Swap-in plan for AWS later" table (the decided Phase 2+ path is not an AWS migration — it's tiered Hetzner VPS via Coolify; AWS Lambda is only an accepted narrow exception for stateless glue work, not a target architecture). This doc also didn't originally cover admin MFA, dependency scanning, CI/CD, or booking-concurrency — those are addressed below and cross-referenced to where the canonical decision lives, rather than duplicated.

Phase 0 (this week): In-app messaging only

Why: quickest privacy win; no phone/email exposure.

Stack

  • FastAPI + JWT
  • Postgres (messages, threads — JSONB for the flexible Message/Thread metadata, per the decided Postgres-only datastore; this doc originally said MongoDB, superseded)
  • Redis (rate limit + short-lived locks)
  • WebSocket (Starlette) for realtime

Models (Pydantic)

class Thread(BaseModel):
    id: str
    consumer_id: str
    provider_id: str
    created_at: datetime
    last_msg_at: datetime
    status: Literal["open","closed","blocked"]

class Message(BaseModel):
    id: str
    thread_id: str
    sender_id: str
    kind: Literal["text","image","file","voice"]
    body: str | None = None          # text/plain
    media_url: AnyUrl | None = None  # S3 later
    created_at: datetime
    e2ee: bool = False               # flip to True when you add E2EE

Endpoints

  • POST /threads (or auto-create on first message)
  • GET /threads?role=provider|consumer&limit=...
  • GET /threads/{id}/messages?cursor=...
  • POST /threads/{id}/messages (text + optional media)
  • WS /ws/threads/{id} (subscribe)
  • Admin/abuse: POST /threads/{id}/block, POST /threads/{id}/close

Security must-haves

  • Per-IP and per-user rate limits (e.g., 20 msg/min)
  • Virus scan uploads (temporary store → scan → promote)
  • Server-side content filters (links, executables)
  • Audit log for every message/meta change

Phase 0.5 (next): Email relay (no AWS yet)

Use Postmark or SendGrid first; both have inbound parse webhooks.

Alias mapping

class EmailAlias(BaseModel):
    alias: EmailStr            # e.g., [email protected]
    thread_id: str
    consumer_id: str | None
    provider_id: str | None
    expires_at: datetime
    status: Literal["active","expired","revoked"]

Flow

  1. Issue alias per thread & role (consumer→provider, provider→consumer).
  2. Outbound: Localz sends via provider alias; reply-to is the alias.
  3. Inbound webhook → verify signature → look up alias → persist as Message(kind="text") → fan-out to app/WebSocket.

FastAPI webhook skeleton

@app.post("/webhooks/email/inbound")
async def inbound_email(req: Request):
    payload = await req.form()
    sig = req.headers.get("X-Provider-Signature")
    verify(sig, payload)                               # HMAC
    alias = payload["to"]
    m = map_alias(alias)                               # EmailAlias
    save_message(thread_id=m.thread_id,
                 sender_id=resolve_sender(payload, m),
                 body=payload["text"])
    notify_ws(thread_id=m.thread_id)
    return {"ok": True}

Additions

  • TTL aliases (e.g., 14 days after last activity)
  • SPF/DKIM/DMARC on localz.app
  • Outbound rate limits, bounce handling, abuse flags

Phase 1: Proxy calls (ship fast with Twilio)

Simpler than Chime to start; replace later without API changes.

Call session model

class CallSession(BaseModel):
    id: str
    thread_id: str
    proxy_number: str       # Twilio number
    consumer_number_last4: str
    provider_number_last4: str
    state: Literal["provisioned","active","ended","expired"]
    expires_at: datetime
    recording_url: AnyUrl | None

Flow

  1. POST /threads/{id}/call/session → provision (choose/allocate proxy DID).
  2. Return a tel: link to the proxy, not the real number.
  3. Twilio hits your voice webhook; you lookup the thread and bridge to the other party.
  4. Optional: record with voice consent.

Voice webhook (TwiML via FastAPI)

@app.post("/voice/bridge")
def bridge_call(form: dict = Depends(parse_twilio_form)):
    sess = find_session_by_proxy(form["To"])
    target = resolve_other_party(sess, caller=form["From"])
    say_consent = "<Say>For quality, this call may be recorded.</Say>"
    return Response(
        f"<Response>{say_consent}<Dial record='record-from-answer' timeout='20'>{target}</Dial></Response>",
        media_type="application/xml"
    )

Hardening

  • Time-based routing (business hours → voicemail/IVR)
  • Max duration caps, per-user/day call caps
  • Recording storage with access controls
  • Number pooling + reuse TTL to control costs

Phase 2: Harden & unify

  • Moderation: heuristic + ML (block phone/email leakage in chat)
  • User controls: per-thread mute, block, export transcript
  • Compliance: consent logs for recording, retention policies
  • Observability: message/call delivery metrics, alerting

Access control, secrets, and CI/CD for this build (previously missing from this doc)

These weren't in the original phase plan above — the messaging/proxy-call build touches admin tooling (block/close threads, recording access) and secrets (Twilio, Postmark/SendGrid, DB credentials) directly enough that it needs its own explicit coverage rather than assuming it's handled elsewhere.

  • Admin MFA: anyone with access to POST /threads/{id}/block-equivalent admin endpoints, or to recording storage, authenticates with MFA — this is the same requirement as Stack-Finalization-Draft.md's "Access, secrets & audit" section, not a separate policy for this subsystem. Don't stand up a second admin auth path for messaging/calls that skips it.
  • Secrets rotation (fixing this doc's original vagueness — it only said ".env now → AWS Secrets Manager later" with no actual rotation policy): Twilio auth tokens, Postmark/SendGrid API keys, and the HMAC signing secret used to verify inbound email webhooks (verify(sig, payload) in the Phase 0.5 skeleton above) must be rotatable on demand and rotated immediately on any suspected exposure — same policy as the canonical one in Stack-Finalization-Draft.md, applied here specifically because this subsystem has more externally-facing secrets (webhook signing, third-party API keys) than most of the app.
  • Dependency scanning: this subsystem pulls in third-party SDKs (Twilio, Postmark/SendGrid clients) and handles untrusted inbound content (email webhooks, uploaded media) — exactly the kind of surface dependency scanning is for. Run it in CI (see below), not as a manual occasional check — a pip-audit/safety step (or GitHub's Dependabot alerts, already free) on every dependency bump is enough at this scale; no need for a paid SCA tool yet.
  • CI/CD: this doc's "unit tests for alias resolution, rate limiting, and webhook verification" (see "What to do this week" below) should run in the same CI pipeline decided in Stack-Finalization-Draft.md's "CI/CD and database migrations" section — gating merges, not a separate ad-hoc test run. The webhook signature verification (verify(sig, payload)) is a good candidate for the first test written here — it's the one piece of this subsystem an attacker can hit directly from the internet.

Not covered by this doc, on purpose: booking-slot concurrency (double-booking prevention) is a different subsystem entirely — messaging/calls don't touch the scheduler. That mechanism (atomic hold via a booking_holds table or advisory lock) lives in Localz-Strategy-and-Architecture-Note.md §6's implementation note, with the correctness requirement itself in Non-functional-Requirements.md §Correctness. Flagged here only so this doc isn't mistaken for the complete security picture — it covers the communication/proxy surface, not the transaction surface.

Production stack alignment (supersedes the old "Swap-in plan for AWS later" table)

The table below originally planned a "start with 3rd-party, later drop in AWS" path. That's no longer the decided direction — Stack-Finalization-Draft.md settled on staying VPS-based (Hetzner + Coolify) rather than migrating to AWS, with AWS Lambda accepted only as a narrow exception for stateless glue code. This table now shows what this subsystem actually runs on, matching that decision:

Capability Start now (3rd-party) Production (decided stack — not AWS)
In-app messaging FastAPI WS + Redis Same — FastAPI WS + Redis, self-hosted on the Hetzner app tier
Media storage Local disk → S3/R2 soon S3 or Cloudflare R2 (decide-later, per Stack-Finalization-Draft)
Email relay Postmark/SendGrid Resend/Postmark (decided sending provider) — Listmonk stays self-hosted for campaign/newsletter UI only, not this relay
Proxy voice/SMS Twilio Proxy/Voice Twilio stays permanent — also now the decided SMS provider (OTP, booking reminders), not just voice
Rate limiting Redis Self-hosted Redis or Upstash (conditional — see Stack-Finalization-Draft's managed-service criteria)
Queues / jobs RQ/Celery/Arq Celery + Redis broker (decided) — with acks_late/idempotency per the Data durability guarantee, since inbound-webhook processing is exactly the kind of in-flight work that guarantee covers
Secrets .env now .env/EnvironmentFile pattern (per the Hermes VPS deploy guide) now, real secrets manager only once there's more than one person needing access — see "Access control, secrets, and CI/CD" above
Audit logs Mongo collection Postgres audit_log/admin_actions table, feeding the self-hosted Grafana/Loki observability stack

Minimal code to get you moving

Rate limit (Redis)

def allow(action_key: str, user_id: str, limit: int, window_s: int) -> bool:
    key = f"rl:{action_key}:{user_id}"
    with redis.pipeline() as p:
        p.incr(key)
        p.expire(key, window_s, nx=True)
        c, _ = p.execute()
    return int(c) <= limit

Cache-aside for thread list

def list_threads(user_id: str):
    key = f"threads:{user_id}"
    data = redis.get(key)
    if data: return json.loads(data)
    rows = db.threads.find({"$or":[{"consumer_id":user_id},{"provider_id":user_id}]})\
                     .sort("last_msg_at",-1).limit(50)
    out = serialize(rows)
    redis.setex(key, 30, json.dumps(out))  # 30s
    return out

What to do this week (10 hrs)

  1. Scaffold endpoints + WS; ship Phase 0 messaging.
  2. Add Redis rate limits + basic moderation + audit log.
  3. Integrate Postmark (outbound) + inbound webhook (Phase 0.5 start).
  4. Create alias TTL job + admin UI for blocks.
  5. Write unit tests for alias resolution, rate limiting, and webhook verification.
Hermes Wiki