Backend architecture

This is the engineering reference for the Lockstep VM Python backend. It is prescriptive: follow the conventions here when adding routes, services, or integrations. Everything below is verified against the code under webapp/backend/app.

The backend is an async FastAPI application backed by MongoDB through Motor (the async driver). There is no ORM — collections are accessed directly as AsyncIOMotorDatabase attributes. Every document is scoped by a tenant_id string, and every query must filter on it.

For the collection shapes these services read and write, see the Data model. For the full route list, see the API reference.

Directory structure

webapp/backend/app/
├── main.py                     # FastAPI app, lifespan, middleware, env-seeding
├── core/                       # Cross-cutting infrastructure
│   ├── config.py               # Pydantic Settings (env → typed settings singleton)
│   ├── database.py             # Motor client lifecycle + get_database()
│   ├── security.py             # get_current_user, JWT helpers, password hashing
│   ├── better_auth.py          # Better Auth session-cookie validation
│   ├── rate_limit.py           # slowapi limiter + key function
│   └── uploads.py              # validate_upload(), sanitize_filename()
├── api/
│   └── v1/
│       ├── __init__.py         # api_router — mounts every endpoint module
│       └── endpoints/          # One module per surface (thin HTTP layer)
│           ├── auth.py
│           ├── inventory.py
│           ├── vulnerabilities.py
│           ├── dashboards.py
│           ├── intune.py  sentinelone.py  connectwise.py  rapid7.py  entra.py ...
│           └── ...
├── services/                   # Reusable business logic (no HTTP concerns)
│   ├── query_builder.py        # Structured Group/Rule → Mongo filter
│   ├── dashboard_service.py    # Dashboard + widget CRUD
│   ├── inventory_rollup.py
│   ├── widget_data_service.py
│   ├── intune_client.py  entra_client.py  connectwise_automate_client.py
│   ├── csv_providers/          # base.py + providers.py
│   └── metrics/                # core_metrics.py
└── models/                     # Pydantic models (base, inventory, vulnerability, user, report)

Two layers do the work:

Layer	Location	Responsibility
Endpoints	api/v1/endpoints/	Thin HTTP adapters: parse/validate input, resolve current_user/db via DI, call services or query Mongo directly, shape the response.
Services	services/	Reusable logic with no FastAPI imports: query building, CRUD helpers, external integration clients, metrics. Take db and tenant_id as plain arguments.

Rule of thumb: an endpoint owns the request/response contract; a service owns logic that could be called from more than one endpoint or a background task.

FastAPI app + lifespan

main.py constructs the app and wires an async lifespan context manager that owns startup/shutdown.

@asynccontextmanager
async def lifespan(app: FastAPI):
    await connect_to_mongo()          # open the Motor client
    await _seed_intune_config()       # env → MongoDB integration seeding
    await _seed_sentinelone_config()
    await _seed_connectwise_config()
    await _seed_rapid7_config()
    await _seed_connectwise_automate_config()
    await _seed_entra_config()
    yield                             # app serves requests here
    await close_mongo_connection()    # close the Motor client

app = FastAPI(
    title=settings.PROJECT_NAME,
    version=settings.VERSION,
    lifespan=lifespan,
    docs_url="/docs", redoc_url="/redoc", openapi_url="/openapi.json",
)

Middleware and the router are registered on the module-level app, in order:

1. Rate limiting — app.state.limiter = limiter, a RateLimitExceeded handler, and SlowAPIMiddleware (see Rate limiting).
2. CORS — origins from settings.cors_origins_list, allow_credentials=True (required so Better Auth session cookies work), a tightened method/header allowlist, and Content-Disposition exposed for file downloads.
3. Router — app.include_router(api_router, prefix=settings.API_V1_PREFIX) mounts everything under /api/v1.

Two unauthenticated probes live directly on app: GET / (name/version) and GET /health. A catch-all @app.exception_handler(Exception) logs the traceback and returns a 500 JSON body {detail, type, path}.

Startup env seeding

Each _seed_* coroutine adopts integration credentials from environment variables into MongoDB once, for the "default" tenant. The pattern is identical across integrations and is the convention for any new integration:

async def _seed_sentinelone_config() -> None:
    if not (settings.SENTINELONE_API_TOKEN and settings.SENTINELONE_CONSOLE_URL):
        return                                   # nothing configured → no-op
    db = get_database()
    if await db.sentinelone_config.find_one({"tenant_id": "default"}):
        return                                   # already seeded → don't clobber
    await db.sentinelone_config.insert_one({
        "tenant_id": "default",
        "api_url": settings.SENTINELONE_CONSOLE_URL,
        "api_token": settings.SENTINELONE_API_TOKEN,
        "enabled": True,
        "sync_status": None, "last_sync": None,
        "updated_at": datetime.utcnow(),
    })

Key invariants for seeders:

• Guard on missing env — return early if the required settings are blank.
• Idempotent — check for an existing {"tenant_id": "default"} doc and skip if present, so a restart never overwrites config a user edited in the UI.
• Never raise — wrap the body in try/except and log; a seed failure must not stop the app from booting.
• The Entra seeder reuses the Intune Azure app registration (same tenant/client/cert), only requiring different Graph scopes.

The api_router registration pattern

api/v1/__init__.py is the single place every endpoint surface is mounted. It imports each module, creates one APIRouter, and include_routers each with a URL prefix and OpenAPI tag.

from app.api.v1.endpoints import auth, inventory, vulnerabilities, dashboards, ...

api_router = APIRouter()

api_router.include_router(auth.router,        prefix="/auth",            tags=["auth"])
api_router.include_router(inventory.router,   prefix="/inventory",       tags=["inventory"])
api_router.include_router(dashboards.router,  prefix="/dashboards",      tags=["dashboards"])
# ...one line per surface

Combined with the app-level prefix, a route defined as @router.get("/uploads") in inventory.py is served at /api/v1/inventory/uploads.

Note two modules can share a prefix: both scan.router and vulnerabilities.router mount under /vulnerabilities.

Adding a new surface

1. Create app/api/v1/endpoints/<name>.py that defines router = APIRouter() and its routes.
2. Add <name> to the import line in app/api/v1/__init__.py.
3. Add api_router.include_router(<name>.router, prefix="/<name>", tags=["<name>"]).

That is the whole wiring — there is no decorator scanning or auto-discovery.

Dependency injection

Endpoints obtain their two universal dependencies through FastAPI Depends. This is the canonical route signature — copy it:

from fastapi import APIRouter, Depends
from motor.motor_asyncio import AsyncIOMotorDatabase
from app.core.security import get_current_user
from app.core.database import get_database

router = APIRouter()

@router.get("/uploads")
async def list_uploads(
    current_user: dict = Depends(get_current_user),
    db: AsyncIOMotorDatabase = Depends(get_database),
):
    cursor = db.inventory_uploads.find({"tenant_id": current_user["tenant_id"]})
    return {"uploads": await cursor.to_list(length=100)}

get_database

Defined in core/database.py. A tiny synchronous function returning the process-wide AsyncIOMotorDatabase opened at startup:

def get_database() -> AsyncIOMotorDatabase:
    return db.db

Motor manages its own connection pool, so a single shared client is correct — do not open a new client per request.

get_current_user

Defined in core/security.py. Resolves the caller from one of two auth sources, in order, and returns a normalized identity dict:

{"user_id": str, "email": str, "tenant_id": str, "role": str, "source": str}

1. Better Auth session cookie (primary, browser traffic) — if the Cookie header contains session_token, validate_session_cookie() forwards the raw cookie to the frontend’s /api/auth/get-session endpoint over the internal Docker network and normalizes the response. Missing tenantId/role default to "default"/"user".
2. Bearer JWT (legacy / programmatic) — otherwise, if an Authorization: Bearer <jwt> header is present, decode it with settings.JWT_SECRET and read sub, email, tenant_id, role from the claims.

If neither resolves, it raises 401. Downstream code only ever sees the normalized dict, so routes never branch on auth source.

get_current_admin_user wraps get_current_user and raises 403 unless role == "admin". Use it as the dependency on admin-only routes:

from app.core.security import get_current_admin_user

@router.delete("/tenant/{tid}")
async def delete_tenant(current_user: dict = Depends(get_current_admin_user), ...):
    ...

Settings & config

core/config.py defines a single Pydantic Settings (from pydantic-settings) loaded from environment variables and an optional .env, exported as the module-level singleton settings. Import it anywhere: from app.core.config import settings.

class Settings(BaseSettings):
    model_config = SettingsConfigDict(env_file=".env", case_sensitive=True)

    PROJECT_NAME: str = "Vulnerability Management API"
    API_V1_PREFIX: str = "/api/v1"
    MONGODB_URL: str            # required — no default
    DATABASE_NAME: str = "vulnmgmt"
    JWT_SECRET: str             # required — no default
    JWT_ALGORITHM: str = "HS256"
    CORS_ORIGINS: str = "http://localhost:4321,http://localhost:3000"
    MAX_UPLOAD_SIZE: int = 100_000_000
    UPLOAD_DIR: str = "./uploads"
    RATE_LIMIT_PER_MINUTE: int = 60
    ENVIRONMENT: str = "development"
    # per-integration credentials: INTUNE_*, SENTINELONE_*, RAPID7_*, CW_AUTOMATE_* ...

Notable fields:

Setting	Default	Purpose
MONGODB_URL	(required)	Motor connection string
DATABASE_NAME	vulnmgmt	Mongo database name
JWT_SECRET	(required)	Signs/verifies legacy JWTs
API_V1_PREFIX	/api/v1	Mounted prefix for api_router
CORS_ORIGINS	localhost:4321,3000	Comma-separated; parsed by cors_origins_list
MAX_UPLOAD_SIZE	100 MB	Upload ceiling enforced in uploads.py
RATE_LIMIT_PER_MINUTE	60	Global slowapi default

CORS_ORIGINS is stored as a comma-separated string and exposed as a list via the cors_origins_list property. Integration credential fields (INTUNE_*, SENTINELONE_*, RAPID7_*, CW_AUTOMATE_*) all default to "" and drive the startup seeders.

MongoDB access with Motor

core/database.py holds the client lifecycle in a small holder object:

class Database:
    client: Optional[AsyncIOMotorClient] = None
    db: Optional[AsyncIOMotorDatabase] = None

db = Database()

async def connect_to_mongo():
    db.client = AsyncIOMotorClient(settings.MONGODB_URL)
    db.db = db.client[settings.DATABASE_NAME]

async def close_mongo_connection():
    if db.client:
        db.client.close()

Conventions when reading/writing Mongo:

• Access collections as attributes of the injected db: db.inventory_uploads, db.applications, db.<collection>. There is no collection registry.

• All operations are awaited: await db.coll.find_one(...), await db.coll.insert_one(...), await db.coll.update_one(...).

• For cursors, always bound the result: await db.coll.find(query).sort(...).to_list(length=N).

• Upsert on natural keys for idempotent ingestion. Example from inventory processing:

  await db.inventory_items.update_one(
      {"tenant_id": tenant_id, "application_name": name,
       "device_name": device, "version": version},
      {"$set": item_dict},
      upsert=True,
  )

• Convert _id to str(result.inserted_id) before returning; use bson.ObjectId(id) to look documents up by id.

Multi-tenancy

Tenancy is enforced by convention, not middleware: every document carries a tenant_id string, and every query filters on it. The tenant comes from current_user["tenant_id"] (Better Auth tenantId, JWT claim, or the "default" fallback).

Rules to follow on every data-access path:

• Reads include tenant_id in the filter:

  await db.inventory_uploads.find({"tenant_id": current_user["tenant_id"]})

• Writes stamp tenant_id (plus created_at/updated_at) into the document.
• Id lookups combine the id with the tenant as defense-in-depth so one tenant can never fetch another’s document by guessing an id:

  await db.scans.find_one({"_id": scan_object_id, "tenant_id": current_user["tenant_id"]})

• Service functions take tenant_id as an explicit leading argument and thread it into every query — never read the request inside a service.
• Uploaded files are namespaced on disk under the tenant: Path(settings.UPLOAD_DIR) / current_user["tenant_id"].
• Integration config docs are keyed on {"tenant_id": ...}; env seeding targets the literal "default" tenant.

Endpoints vs services split

Services never import FastAPI. They expose plain async functions whose first parameters are the db and the tenant_id, keeping them callable from endpoints and background tasks alike. dashboard_service.py is the reference shape:

# services/dashboard_service.py — signatures
async def list_dashboards(db: AsyncIOMotorDatabase, tenant_id: str, owner_user_id: str) -> List[dict]: ...
async def get_dashboard(db: AsyncIOMotorDatabase, tenant_id: str, dashboard_id: str) -> Optional[dict]: ...
async def create_dashboard(db, tenant_id, ...) -> dict: ...
async def ensure_system_dashboards(db: AsyncIOMotorDatabase, tenant_id: str) -> dict: ...

The endpoint stays thin — resolve DI, delegate, shape the response:

from app.services import dashboard_service

@router.get("/")
async def list_dashboards(
    current_user: dict = Depends(get_current_user),
    db: AsyncIOMotorDatabase = Depends(get_database),
):
    return await dashboard_service.list_dashboards(
        db, current_user["tenant_id"], current_user["user_id"],
    )

Other service categories follow the same rule:

• query_builder.py — validates a nested Group/Rule payload against a caller-supplied field allowlist and returns a Mongo filter document, so the UI can never reference arbitrary field paths.
• Integration clients (intune_client.py, entra_client.py, connectwise_automate_client.py, csv_providers/) — wrap external APIs and normalize their responses.
• metrics/, widget_data_service.py, inventory_rollup.py — aggregation/rollup logic used by dashboards and reports.

Background tasks

Long-running work (file processing, integration syncs) uses FastAPI’s BackgroundTasks. The endpoint persists a status document, schedules the worker, and returns immediately with an id the client can poll.

@router.post("/upload", status_code=201)
async def upload_inventory(
    background_tasks: BackgroundTasks,
    file: UploadFile = File(...),
    current_user: dict = Depends(get_current_user),
    db: AsyncIOMotorDatabase = Depends(get_database),
):
    content = await validate_upload(file, kind="csv")        # size/ext/MIME checks
    # ... write file to the tenant's upload dir, insert an "uploaded" status doc ...
    result = await db.inventory_uploads.insert_one(upload_doc)
    upload_id = str(result.inserted_id)

    background_tasks.add_task(
        process_inventory_file,           # module-level async worker
        str(file_path), upload_id,
        current_user["tenant_id"],        # tenancy passed explicitly
        db,                               # shared Motor db handed to the task
    )
    return {"upload_id": upload_id, "status": "processing"}

Conventions for background workers:

• The worker is a module-level async function taking tenant_id and db explicitly (no request access outside the handler).
• It writes progress back into a status document — status transitions like uploaded → processing → processed / failed with an error_message. Poll endpoints read that doc.
• Wrap the body in try/except and record status: "failed" on error — an unhandled exception in a background task is not surfaced to the client.
• BackgroundTasks run in the same process after the response is sent; they are not a durable queue. Used across intune, sentinelone, connectwise_automate, rapid7, entra, news_feeds, subscriptions, csv_sources, reports_enhanced, and scan.

Rate limiting

core/rate_limit.py configures a slowapi Limiter. The key function prefers the JWT sub (so users behind one NAT IP don’t starve each other) and falls back to the client IP for unauthenticated routes:

limiter = Limiter(
    key_func=_rate_limit_key,
    default_limits=[f"{settings.RATE_LIMIT_PER_MINUTE}/minute"],
    headers_enabled=True,
)

The global default applies to every route. Sensitive routes (login, register) tighten it with a per-route @limiter.limit("...") decorator.

Upload validation

Any endpoint accepting an UploadFile must run it through core/uploads.validate_upload() before persisting. It enforces three layers and returns the raw bytes so callers don’t re-read the stream:

from app.core.uploads import validate_upload, sanitize_filename

content = await validate_upload(file, kind="csv")           # → bytes, or raises 400
safe_name = sanitize_filename(file.filename or "upload.csv")

Check	Enforced by	Failure
Extension	ACCEPTED_KINDS[kind]["extensions"] allowlist	400 Upload rejected
Size	max_size arg or settings.MAX_UPLOAD_SIZE	400 Upload rejected
Content type	libmagic MIME sniff vs. mime_prefixes	400 Upload rejected

Supported kind values are "csv", "excel", "pem", and "json". If python-magic/libmagic is unavailable at runtime, MIME sniffing degrades to a no-op (logged) while extension and size checks still run. sanitize_filename() strips path separators and control characters — always run user-supplied names through it before building a disk path.

Request lifecycle

End-to-end path of an authenticated request (GET /api/v1/inventory/uploads):

1. CORS — CORSMiddleware handles preflight/headers; credentialed requests are allowed only from settings.cors_origins_list.
2. Rate limit — SlowAPIMiddleware keys the request by JWT sub or client IP and enforces the per-minute limit; over-limit → 429.
3. Routing — matched to the inventory.router route mounted at /api/v1/inventory by api_router.
4. Dependency resolution — get_current_user resolves identity (Better Auth cookie → JWT), yielding the normalized {user_id, email, tenant_id, role} dict; get_database returns the shared Motor db. A failure here short-circuits to 401/403.
5. Handler — the endpoint queries Mongo (or a service) scoped by current_user["tenant_id"], optionally schedules a BackgroundTasks worker, and returns a JSON-serializable value.
6. Response — FastAPI serializes it; background tasks (if any) run after the response is flushed.
7. Errors — HTTPExceptions become their status/detail; anything uncaught hits the global Exception handler → logged traceback + 500 {detail, type, path}.

Conventions checklist

When adding backend code:

• New surface → module in endpoints/ exposing router, imported and include_routered in api/v1/__init__.py.
• Every data route depends on get_current_user and get_database.
• Every query filters on tenant_id; every write stamps tenant_id + timestamps.
• Reusable logic lives in services/, takes (db, tenant_id, ...), and imports no FastAPI.
• Uploads go through validate_upload(); filenames through sanitize_filename().
• Long work runs via BackgroundTasks with a pollable status document and try/except failure recording.
• New integrations add an idempotent, guard-early, never-raise _seed_* seeder to the lifespan.

See Docker & local setup for running the backend end to end, and the API reference for the concrete route surface.