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 Security — the 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
- Typer documentation — "Tutorial - User Guide" (first steps + Commands + Arguments/Options)
(~40 min) — how a type-annotated function becomes a CLI; read how parameter types drive parsing and
--help. This is the "types → interface" idea you'll reuse on the API side.
FastAPI — the HTTP surface
- FastAPI documentation — "Tutorial - User Guide" (First Steps + Request Body) (~45 min) — the key section: a pydantic model as a request-body parameter gives you automatic validation, JSON parsing, and OpenAPI docs. This is the direct payoff of your M2 models.
- FastAPI — "Response Model" (~15 min) — typing the return as a pydantic model so the response is validated and documented too; closes the loop with the request side.
The shared-core pattern (why not to duplicate)
- FastAPI — "Bigger Applications - Multiple Files" (~20 min) — how to structure a project so routers/adapters import a shared core module; read it as the blueprint for keeping logic out of the surfaces.
- Cosmic Python (Percival & Gregory) — "A Brief Interlude: On Coupling and Abstractions" (~25 min) — the why behind thin adapters over a stable core; the entrypoints-as-thin-shells argument, vendor-neutral and directly applicable here.
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.
typerandFastAPIshare a shape — both derive their interface from your type annotations.- FastAPI validates request bodies against your
AlertEventEVE model — a bad EVE line becomes a422, 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 /triagerequest 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).