mirror of
https://github.com/computerim/impactflow-discovery.git
synced 2026-08-27 06:00:35 +00:00
Joel Salmon
b8f176bb31
Discovery service (pre-existing): FastAPI + async SQLAlchemy + Alembic +
SQLite + Anthropic, with a five-prompt static UI that produces an Enneagram
+ Ikigai profile.
Auth implementation (this change set) follows
Impact_Flow_Auth_Plan_OAuth.html, adapted to the discovery_conversation /
discovery_profile schema:
- app/auth.py: Google OAuth registration, JWT issue/decode, dual-auth
dependency (Bearer JWT or X-API-Key), refresh-token hashing, domain
allow-list, synthetic api-key-admin user
- app/tracking.py: ActivityTrackingMiddleware + log_activity helper;
tags machine-to-machine calls source=mcp
- app/routers/auth.py: /api/auth/{login,callback,refresh,logout},
/api/me, /api/me/{stats,sessions,sessions/{id}}
- app/routers/activity.py: /api/activity, /api/activity/summary,
/api/admin/activity, plus prune_old_activity (90-day retention)
- app/routers/discovery.py: every route now user-scoped via the auth
dependency; /discovery/profile/{user_id} -> /discovery/profile/me
- alembic/versions/002_add_auth.py: users, refresh_tokens, activity_log
- tests/test_auth.py: 8 tests covering 401 paths, X-API-Key admin
resolution, JWT round-trip, admin gating, domain allow-list
- README.md: Authentication section, expanded env-var table, updated
data-model and API-reference tables
- .env.example: new GOOGLE_*, JWT_*, IMPACTFLOW_API_KEY, CORS_*,
ALLOWED_EMAIL_DOMAINS placeholders
- .gitignore: also exclude data/*.log
Tests: 19/19 pass (11 pre-existing + 8 new). smoke_test.py exercises the
full discovery flow under X-API-Key plus 401 paths, OAuth login redirect,
activity logging, and /api/me/stats.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
112 lines
3.3 KiB
Python
112 lines
3.3 KiB
Python
"""Activity-log helpers + middleware.
|
|
|
|
Middleware fires after every request and writes one ActivityLog row when
|
|
the request was authenticated and succeeded (status < 400). We classify the
|
|
verb (GET -> view / POST -> create / etc.) and bucket the URL path into a
|
|
coarse resource name so the dashboard can group "everything about
|
|
conversations" without parsing URLs at query time.
|
|
"""
|
|
import json
|
|
import uuid
|
|
from datetime import datetime, timezone
|
|
from typing import Optional
|
|
|
|
from fastapi import Request
|
|
from starlette.middleware.base import BaseHTTPMiddleware
|
|
|
|
from app.database import AsyncSessionLocal
|
|
from app.models import ActivityLog
|
|
|
|
_METHOD_ACTION = {
|
|
"GET": "view",
|
|
"POST": "create",
|
|
"PUT": "update",
|
|
"PATCH": "update",
|
|
"DELETE": "delete",
|
|
}
|
|
|
|
# Paths whose activity is not worth recording (noisy / no user signal).
|
|
_SKIP_PATH_PREFIXES = (
|
|
"/health",
|
|
"/static",
|
|
"/favicon.ico",
|
|
"/openapi.json",
|
|
"/docs",
|
|
"/redoc",
|
|
"/api/auth/login",
|
|
"/api/auth/callback",
|
|
)
|
|
|
|
|
|
def _method_to_action(method: str) -> str:
|
|
return _METHOD_ACTION.get(method.upper(), "other")
|
|
|
|
|
|
def _path_to_resource(path: str) -> str:
|
|
parts = [p for p in path.strip("/").split("/") if p]
|
|
if not parts:
|
|
return "root"
|
|
if parts[0] == "api" and len(parts) > 1:
|
|
return parts[1]
|
|
return parts[0]
|
|
|
|
|
|
async def log_activity(
|
|
*,
|
|
user_id: str,
|
|
action: str,
|
|
resource: str,
|
|
resource_id: Optional[str] = None,
|
|
metadata: Optional[dict] = None,
|
|
source: str = "web",
|
|
ip_address: Optional[str] = None,
|
|
user_agent: Optional[str] = None,
|
|
) -> None:
|
|
"""Write one row. Uses its own session because the request's session is
|
|
already closed by the time tracking middleware sees the response."""
|
|
row = ActivityLog(
|
|
id=str(uuid.uuid4()),
|
|
user_id=user_id,
|
|
action=action,
|
|
resource=resource,
|
|
resource_id=resource_id,
|
|
metadata_json=json.dumps(metadata) if metadata else None,
|
|
source=source,
|
|
ip_address=ip_address,
|
|
user_agent=(user_agent or "")[:1000] or None,
|
|
created_at=datetime.now(timezone.utc),
|
|
)
|
|
async with AsyncSessionLocal() as db:
|
|
db.add(row)
|
|
await db.commit()
|
|
|
|
|
|
class ActivityTrackingMiddleware(BaseHTTPMiddleware):
|
|
async def dispatch(self, request: Request, call_next):
|
|
response = await call_next(request)
|
|
|
|
# Only log authenticated, successful, non-noisy requests.
|
|
if response.status_code >= 400:
|
|
return response
|
|
path = request.url.path
|
|
if any(path.startswith(p) for p in _SKIP_PATH_PREFIXES):
|
|
return response
|
|
user = getattr(request.state, "user", None)
|
|
if user is None:
|
|
return response
|
|
|
|
source = "mcp" if request.headers.get("x-api-key") else "web"
|
|
try:
|
|
await log_activity(
|
|
user_id=user.id,
|
|
action=_method_to_action(request.method),
|
|
resource=_path_to_resource(path),
|
|
source=source,
|
|
ip_address=request.client.host if request.client else None,
|
|
user_agent=request.headers.get("user-agent"),
|
|
)
|
|
except Exception:
|
|
# Tracking failures must never break the underlying request.
|
|
pass
|
|
return response
|