Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions .env.example
Original file line number Diff line number Diff line change
Expand Up @@ -18,3 +18,8 @@ API_PORT=8000

# CORS — comma-separated list of allowed origins
CORS_ORIGINS=http://localhost:3000,http://localhost:8000

# App API bearer token — gates /api/app/* (companion app + scheduler).
# Distinct from any MCP token. Generate with: python -m scripts.generate_token
# Leave unset in dev to skip auth (a startup warning will be logged).
# BRAIN3_APP_BEARER_TOKEN=
92 changes: 92 additions & 0 deletions app/auth.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,92 @@
# BRAIN 3.0 — AI-powered personal operating system for ADHD
# Copyright (C) 2026 L (WilliM233)
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as
# published by the Free Software Foundation, either version 3 of the
# License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.

"""Bearer token authentication for the App API surface (``/api/app/*``).

Mirrors the pattern in ``brain3-mcp/mcp/auth.py``: a Starlette
``BaseHTTPMiddleware`` that validates ``Authorization: Bearer <token>`` using
``hmac.compare_digest`` for constant-time comparison.

Scoped by path prefix so the existing ``/api/*`` surface (consumed by
``brain3-mcp`` and other internal clients) remains unauthenticated. Only
requests under ``/api/app/`` are gated. The token is sourced from
``settings.APP_BEARER_TOKEN`` (env var ``BRAIN3_APP_BEARER_TOKEN``) and is
intentionally distinct from any MCP token — separate secrets, separate
rotation.
"""

from __future__ import annotations

import hmac
import logging

from fastapi import FastAPI
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
from starlette.responses import JSONResponse, Response

logger = logging.getLogger(__name__)

APP_PATH_PREFIX = "/api/app/"


class AppBearerAuthMiddleware(BaseHTTPMiddleware):
"""Bearer auth on ``/api/app/*``. Pass-through for all other paths.

Token comparison uses ``hmac.compare_digest`` to prevent timing attacks
against the shared secret.
"""

def __init__(self, app, token: str) -> None:
super().__init__(app)
self.token = token

async def dispatch(self, request: Request, call_next) -> Response:
if not request.url.path.startswith(APP_PATH_PREFIX):
return await call_next(request)

auth_header = request.headers.get("authorization", "")
if not auth_header.startswith("Bearer "):
return JSONResponse(
{"error": "Missing bearer token"}, status_code=401,
)

provided = auth_header[len("Bearer ") :]
if not hmac.compare_digest(provided, self.token):
return JSONResponse(
{"error": "Invalid bearer token"}, status_code=401,
)

return await call_next(request)


def install_app_bearer_auth(app: FastAPI, token: str | None) -> None:
"""Mount :class:`AppBearerAuthMiddleware` if ``token`` is set.

When ``token`` is ``None`` or empty, the middleware is **not** mounted
and a startup warning is logged. This matches the existing graceful-
degradation pattern used elsewhere in the app and lets local dev work
without forcing a token in ``.env``. Production deployments must set
``BRAIN3_APP_BEARER_TOKEN`` before exposing the app API.
"""
if not token:
logger.warning(
"BRAIN3_APP_BEARER_TOKEN is not set — /api/app/* endpoints are "
"UNAUTHENTICATED. Set the env var to enable bearer auth before "
"exposing the app API.",
)
return
app.add_middleware(AppBearerAuthMiddleware, token=token)
12 changes: 11 additions & 1 deletion app/config.py
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@

"""Application configuration via Pydantic Settings."""

from pydantic import Field
from pydantic_settings import BaseSettings, SettingsConfigDict


Expand All @@ -33,7 +34,16 @@ class Settings(BaseSettings):

CORS_ORIGINS: str = "http://localhost:3000,http://localhost:8000"

model_config = SettingsConfigDict(env_file=".env", env_file_encoding="utf-8")
# Shared-secret bearer token gating /api/app/* (companion app + scheduler).
# Distinct from any MCP token. Unset in dev → middleware skipped with a
# startup warning. Required in production.
APP_BEARER_TOKEN: str | None = Field(
default=None, validation_alias="BRAIN3_APP_BEARER_TOKEN",
)

model_config = SettingsConfigDict(
env_file=".env", env_file_encoding="utf-8", populate_by_name=True,
)

@property
def database_url(self) -> str:
Expand Down
16 changes: 16 additions & 0 deletions app/main.py
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@
from fastapi.responses import JSONResponse
from sqlalchemy import text

from app.auth import install_app_bearer_auth
from app.config import settings
from app.database import SessionLocal
from app.routers import (
Expand Down Expand Up @@ -57,6 +58,10 @@
allow_headers=["*"],
)

# Bearer auth on /api/app/*. Skipped with a warning when the token is unset
# (dev convenience) — production must set BRAIN3_APP_BEARER_TOKEN.
install_app_bearer_auth(app, settings.APP_BEARER_TOKEN)


@app.get("/health")
def health_check() -> dict:
Expand All @@ -75,6 +80,17 @@ def health_check() -> dict:
)


@app.get("/api/app/health")
def app_health_check() -> dict:
"""Authed probe endpoint for the companion app's validation ping.

Sits behind ``AppBearerAuthMiddleware``. The companion app calls this
during URL+token pairing to confirm both that the server is reachable
and that the bearer token is valid.
"""
return {"ok": True}


app.include_router(domains.router, prefix="/api/domains", tags=["Domains"])
app.include_router(goals.router, prefix="/api/goals", tags=["Goals"])
app.include_router(projects.router, prefix="/api/projects", tags=["Projects"])
Expand Down
59 changes: 59 additions & 0 deletions scripts/generate_token.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,59 @@
#!/usr/bin/env python3
# BRAIN 3.0 — AI-powered personal operating system for ADHD
# Copyright (C) 2026 L (WilliM233)
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as
# published by the Free Software Foundation, either version 3 of the
# License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.

"""Generate a BRAIN 3.0 app-API bearer token.

Prints a URL-safe 48-byte random token to stdout. The companion app and the
scheduler authenticate to ``/api/app/*`` with this token, so it should be
treated as a shared secret. After generating, set::

BRAIN3_APP_BEARER_TOKEN=<value>

in the ``.env`` file (or your deployment secret store), then restart the API.

The token is the shared secret itself — there is no DB write, no user
record, no per-device state. Rotation = generate a new token, update the
env, restart the API. (Per-device tokens and rotation tooling are Phase 3.)

Usage:
python -m scripts.generate_token
"""

from __future__ import annotations

import secrets
import sys

TOKEN_BYTES = 48


def main() -> None:
"""Print a fresh URL-safe token to stdout, with a hint to stderr."""
token = secrets.token_urlsafe(TOKEN_BYTES)
# Token-only on stdout so it captures cleanly:
# TOKEN=$(python -m scripts.generate_token)
print(token)
# Operator hint goes to stderr so it doesn't pollute the captured value.
print(
f"Set BRAIN3_APP_BEARER_TOKEN={token} in your .env file "
"(or secret store) and restart the API to enable /api/app/* auth.",
file=sys.stderr,
)


if __name__ == "__main__":
main()
146 changes: 146 additions & 0 deletions tests/test_app_auth.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,146 @@
# BRAIN 3.0 — AI-powered personal operating system for ADHD
# Copyright (C) 2026 L (WilliM233)
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as
# published by the Free Software Foundation, either version 3 of the
# License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.

"""Tests for the App API bearer-token middleware (``app/auth.py``).

Covers the four acceptance paths from issue #200 — missing header, wrong
token, correct token, env-unset warning — plus the no-regression check that
non-``/api/app/*`` routes stay unauthenticated when the middleware is on.

Each test builds its own ``FastAPI`` app rather than reusing the global
``app.main.app``: that app's middleware is wired up at import time from the
real env, so per-test scenarios (no-token, known-token) need a clean
instance.
"""

from __future__ import annotations

import logging

import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient

from app.auth import AppBearerAuthMiddleware, install_app_bearer_auth

TEST_TOKEN = "test-bearer-token-abc123" # noqa: S105 — fixture, not a real secret


def _app_with_token(token: str) -> FastAPI:
"""Build a fresh app with the middleware mounted to ``token``."""
fresh = FastAPI()
fresh.add_middleware(AppBearerAuthMiddleware, token=token)

@fresh.get("/api/app/health")
def app_health() -> dict:
return {"ok": True}

@fresh.get("/api/notifications/ping")
def notif_ping() -> dict:
return {"pong": True}

return fresh


@pytest.fixture
def authed_client() -> TestClient:
"""TestClient against a fresh app with the bearer middleware enabled."""
return TestClient(_app_with_token(TEST_TOKEN))


# ---------------------------------------------------------------------------
# Acceptance paths — issue #200
# ---------------------------------------------------------------------------

def test_missing_authorization_header_returns_401(authed_client: TestClient) -> None:
"""No Authorization header → 401 with the missing-token error body."""
resp = authed_client.get("/api/app/health")
assert resp.status_code == 401
assert resp.json() == {"error": "Missing bearer token"}


def test_wrong_token_returns_401(authed_client: TestClient) -> None:
"""Authorization header with the wrong secret → 401 invalid-token."""
resp = authed_client.get(
"/api/app/health",
headers={"Authorization": "Bearer not-the-real-token"},
)
assert resp.status_code == 401
assert resp.json() == {"error": "Invalid bearer token"}


def test_correct_token_returns_200(authed_client: TestClient) -> None:
"""Authorization header with the right secret → endpoint runs."""
resp = authed_client.get(
"/api/app/health",
headers={"Authorization": f"Bearer {TEST_TOKEN}"},
)
assert resp.status_code == 200
assert resp.json() == {"ok": True}


def test_unset_token_emits_warning_and_skips_middleware(
caplog: pytest.LogCaptureFixture,
) -> None:
"""``install_app_bearer_auth`` warns and stays a no-op when token is unset."""
fresh = FastAPI()

@fresh.get("/api/app/health")
def app_health() -> dict:
return {"ok": True}

with caplog.at_level(logging.WARNING, logger="app.auth"):
install_app_bearer_auth(fresh, None)

assert any(
"BRAIN3_APP_BEARER_TOKEN is not set" in record.message
and record.levelno == logging.WARNING
for record in caplog.records
), f"Expected unset-token warning; got: {[r.message for r in caplog.records]}"

# Middleware was not mounted: the endpoint serves without an auth header.
with TestClient(fresh) as client:
resp = client.get("/api/app/health")
assert resp.status_code == 200
assert resp.json() == {"ok": True}


# ---------------------------------------------------------------------------
# Regression — issue #200: existing /api/* routes stay unauthenticated
# ---------------------------------------------------------------------------

def test_non_app_paths_pass_through_when_middleware_active(
authed_client: TestClient,
) -> None:
"""Routes outside ``/api/app/*`` are not gated even with the middleware on.

Guards the no-regression criterion: ``/api/notifications/*`` (and every
other existing route consumed by MCP / internal clients) must still
answer without an Authorization header.
"""
resp = authed_client.get("/api/notifications/ping")
assert resp.status_code == 200
assert resp.json() == {"pong": True}


def test_empty_bearer_value_returns_401(authed_client: TestClient) -> None:
"""Header present but value empty → still missing-token (no Bearer prefix)."""
resp = authed_client.get(
"/api/app/health",
headers={"Authorization": ""},
)
assert resp.status_code == 401
assert resp.json() == {"error": "Missing bearer token"}
Loading