Skip to content

Module 06 — Two Surfaces, One Core

Type 7 · Build-&-Operate — stand a validated sift core behind two surfaces (a typer CLI an analyst runs and a FastAPI service a pipeline calls) so the same typed logic ships once and is consumed two ways, without duplicating a line. Go to the hands-on lab →  ·  Cheat sheet →

Last reviewed: 2026-08

Python for Securitythe copilot writes the code in seconds; your edge is the project it writes into and the spec it writes against.

In 60 seconds

A good security tool has to be usable two ways at once: an analyst runs it at a terminal, and a pipeline (a SOAR playbook, another service) calls it over HTTP. The copilot's default when you ask for both is to write the logic twice — once behind typer, once behind FastAPI — and now you have two triage engines that drift apart. The right shape is one core, two surfaces: the pydantic models and the enrich/triage functions you built in M2–M5 are the core, and typer and FastAPI are thin adapters that call into it. The same AlertEvent EVE model validates a CLI-read eve.json line and an HTTP request body. This is the payoff of the whole parse, don't trust spine: FastAPI validates every request against your models for free, because they already exist — a truncated or off-range EVE line becomes a clean 422, not a crash. This module closes Phase 2.

Why this matters

Every mature SOC eventually needs its tooling as a service. The enrichment tool an analyst runs by hand during an investigation is the same logic a detection pipeline needs to call automatically — a SOAR playbook that enriches every indicator on ingest, another microservice that wants a triage verdict before it pages someone. If the CLI and the service are two separate codebases, they will diverge: a scoring tweak lands in one and not the other, and now the analyst and the automation disagree about whether an alert is critical. That divergence is an incident waiting to happen — the automation suppresses what the human would have escalated.

So the operational requirement is: one implementation, reachable two ways. This is a Build-&-Operate problem, not a vulnerability — the thing that bites you is architecture and toil, not a CVE. And it's exactly where the copilot's instinct fails, because "add a CLI" and "add an API" are two separate prompts and the model happily answers each by re-implementing the triage logic inline. Your job is to hold the line that the logic lives in one place and both surfaces are dumb adapters over it.

Objective

Expose sift's existing validated core through two thin adapters — a typer CLI and a FastAPI service — that share the same pydantic models and core functions with zero duplicated business logic; make FastAPI validate request bodies against the AlertEvent EVE model you already own (a bad EVE line → HTTP 422); and prove both surfaces produce identical results from the same EVE input.

The core idea

sift is one core reached two ways: the validated pydantic models and core functions you built in M2–M5 are the tool, and typer and FastAPI are thin adapters that import and delegate — the scoring logic lives in exactly one place, so the analyst's CLI and the pipeline's API can never drift apart.

flowchart TB
    CLI["typer CLI<br/>analyst runs sift triage"] -->|import + delegate| CORE
    API["FastAPI service<br/>pipeline calls POST /triage"] -->|import + delegate| CORE
    subgraph CORE["sift core — one implementation, built M2–M5"]
        M["pydantic models<br/>AlertEvent · TriageResult"]
        F["core functions<br/>triage() · enrich()"]
        M --- F
    end
    CORE -.the scoring logic appears here<br/>and nowhere else.-> CORE

The core already exists — surfaces are adapters, not owners. By the end of M5, sift has a typed core: the pydantic EVE AlertEvent model (M2), a streaming triage layer (M3), an async enricher (M4), and safe tool wrappers (M5), all reachable as plain Python functions like enrich(event) -> AlertEvent and triage(event) -> TriageResult. A surface's only jobs are to (1) get input from somewhere — an eve.json line on argv, or an HTTP body — into your models, (2) call a core function, and (3) render the result back out — to a terminal, or as JSON. The moment a surface contains an if severity > ... scoring decision, you've leaked business logic out of the core, and the two surfaces have started to drift. The discipline: a typer command or a FastAPI endpoint should be a handful of lines that import and delegate.

typer and FastAPI are the same shape — both are "typed function → interface." typer turns a type-annotated function into a CLI: parameter types become argument parsers and --help text. FastAPI turns a type-annotated function into an HTTP endpoint: a pydantic parameter becomes a request-body schema with automatic validation and OpenAPI docs. Once you see that both frameworks derive the interface from your types, the "one core, two surfaces" pattern stops being extra work — you write the typed core once, and each adapter is a decorator plus a delegate call. Both surfaces walk the same path — get input into the model, call one core function, render the result out — they only differ at the two ends:

flowchart LR
    IN1["eve.json line<br/>CLI · on argv"] --> V
    IN2["HTTP body<br/>API · POST /triage"] --> V
    V["validate into<br/>AlertEvent (the M2 model)"] --> C["triage(event)<br/>one core function"]
    C --> R1["JSON to stdout<br/>CLI render"]
    C --> R2["200 + TriageResult<br/>API response"]
typer CLI FastAPI service
Consumer analyst at a terminal pipeline / SOAR playbook, over HTTP
Input an eve.json line on argv a JSON request body
Validation AlertEvent.model_validate_json(...) FastAPI validates the body → clean 422
Interface derived from a typed function → args + --help a typed function → request schema + OpenAPI
Output JSON to stdout JSON HTTP response (TriageResult)
Business logic none — imports the core none — imports the core

FastAPI's pydantic-native validation is the whole spine paying off. This is why we spent M2 building models that reject adversarial input. When you type a FastAPI endpoint's body as event: AlertEvent, FastAPI validates every incoming request against that EVE model before your code runs — a truncated eve.json line, an out-of-range alert.severity, or a missing pinned field becomes a clean 422 with a precise error, not a crash deep in your enricher. The untrusted-input boundary you built for the CLI is now defending your HTTP surface too, for free, because it's the same model. Parse-don't-trust was never about one edge; it was about owning the type that every edge validates against.

flowchart LR
    REQ(["POST /triage<br/>request body — untrusted"]) --> G{"validate against<br/>AlertEvent · the M2 model"}
    G -->|valid EVE record| OK["triage() runs<br/>→ 200 + TriageResult"]
    G -->|truncated / severity ∉ 1..3 / wrong event_type| ERR["422 Unprocessable<br/>before your code runs ✓"]

The through-line — parse-don't-trust at a second edge

The M2 discipline was own the type that validates untrusted input. An HTTP request body is untrusted input too — so typing a FastAPI endpoint's body as event: AlertEvent reuses the exact same model to guard the API edge that guarded the CLI edge. You wrote the boundary once; both surfaces inherit it.

# core.py — the shared core. Already exists from M2–M5. No CLI, no HTTP. Just types + logic.
from .models import AlertEvent, TriageResult   # EVE model (M2) + triage output

def triage(event: AlertEvent) -> TriageResult:  # the one implementation of the logic
    ...
# cli.py — typer adapter. Thin: parse a real EVE line into the model, delegate, render.
import typer
from .core import triage
from .models import AlertEvent

app = typer.Typer()

@app.command()
def run(eve_file: typer.FileText) -> None:
    event = AlertEvent.model_validate_json(eve_file.readline())  # same model validates one eve.json record
    result = triage(event)                                        # same core function
    typer.echo(result.model_dump_json(indent=2))
# api.py — FastAPI adapter. Thin: FastAPI validates the EVE body into the model, delegate, return.
from fastapi import FastAPI
from .core import triage
from .models import AlertEvent, TriageResult

api = FastAPI()

@api.post("/triage")
def triage_endpoint(event: AlertEvent) -> TriageResult:  # FastAPI validates the EVE body against AlertEvent -> 422 on a bad line
    return triage(event)                                  # same core function; identical result to the CLI

Two files, one import triage. The scoring logic appears zero times in either adapter.

Doesn't the async enricher (M4) force async def on the surfaces?

Only where you actually await. FastAPI supports both def and async def endpoints natively, so an endpoint that calls your M4 async enricher is simply async def ... await enrich(...). typer commands are sync, so the CLI wraps the same coroutine in asyncio.run(...). Crucially, the enrichment logic itself still lives once in the core — each surface only chooses how it invokes it.

Go deeper (~2–3 hrs · optional)

The core idea above teaches the one-core-two-surfaces move and the parse-don't-trust-at-a-second-edge payoff, and you can do the lab from it. These links go deeper on each framework and the "thin adapter over a stable core" argument — pull them when a step doesn't click, not as required reading.

Typer — the CLI surface

FastAPI — the HTTP surface

The shared-core pattern (why not to duplicate)

Key concepts

  • One core, two surfaces — the pydantic models + core functions are the tool; CLI and API are adapters.
  • A surface is thin — get input into a model, call a core function, render the result out. No business logic.
  • typer and FastAPI share a shape — both derive their interface from your type annotations.
  • FastAPI validates request bodies against your AlertEvent EVE model — a bad EVE line becomes a 422, not a crash.
  • Zero duplicated logic is the acceptance bar — the scoring/triage code appears exactly once in the repo.

AI acceleration

Ask a copilot to "add a CLI and a REST API to this tool" and watch the failure-class appear: it will scaffold typer and FastAPI independently, and re-implement the triage/scoring logic inside each one — often with subtle differences, because it generated them in two separate passes. The review move is to trace the business logic: it must appear exactly once, in the core, and both adapters must import it. If you see an if-branch that scores an alert inside a @app.command() or an @api.post(), reject it — that's the duplication the module targets. Write the spec as "two thin adapters over the existing core, zero duplicated logic, both surfaces validate against the same models," and review the diff against exactly that.

Check yourself

  • Where does the triage/scoring logic live, and how many times does it appear in the whole repo?
  • When a POST /triage request arrives with a missing required field, what validates it and what does the client get back — and why did you get that for free?
  • If a teammate adds a new severity rule, which files change so the CLI and the API stay in agreement?

Comments

Sign in with GitHub to comment. Choose the type: Feedback (errors or suggestions on this page) · Hints (help for fellow learners — no spoilers) · General (anything else).