Backend Live Coding

-- ======================================================================
-- Q: How do you apply PostgreSQL Row-Level Security for multi-tenancy?
--
-- It is a second line of defense: even if a filter is forgotten in the code, the
-- database does not return another tenant's data. Using set_config with true
-- (local) is important when you use a connection pool, so the tenant does not
-- "stick" to the connection.
-- ======================================================================

ALTER TABLE assets ENABLE ROW LEVEL SECURITY;
ALTER TABLE assets FORCE ROW LEVEL SECURITY;      -- applies to the owner too
CREATE POLICY tenant_isolation ON assets
    USING (tenant_id = current_setting('app.tenant_id')::int);

# In every transaction, from the backend:
session.execute(text("SELECT set_config('app.tenant_id', :t, true)"),
                {'t': str(user.tenant_id)})        # true = only for this transaction

# ======================================================================
# Q: We want an LLM to read free text (e.g. vendor advisories) and produce
# structured fields. How do you build it?
#
# The LLM is just another external, slow and non-deterministic API, so
# everything I know about integrations applies, plus validation of the result. I
# hide the provider behind our own interface so the model or provider can be
# swapped.
# Signs of maturity: structured output with a schema and validation, async in
# Celery with rate limit and retries, a cache keyed by the hash of the input
# (cost), recording the model and prompt version on every result
# (audit/compliance), human review for anything that fails or has low
# confidence, and measuring tokens/cost per tenant.
# ======================================================================

class AdvisoryExtract(BaseModel):
    cve_ids: list[str] = Field(default_factory=list)
    affected_products: list[str]
    severity: Literal['low', 'medium', 'high', 'critical']
    summary: str = Field(max_length=500)

    @field_validator('cve_ids')
    @classmethod
    def valid_cves(cls, v):
        return [c for c in v if re.fullmatch(r'CVE-\d{4}-\d{4,}', c)]

@shared_task(bind=True, autoretry_for=(LLMTimeout, LLMRateLimited),
             retry_backoff=True, max_retries=5, rate_limit='30/m')
def extract_advisory(self, doc_id):
    doc = load_document(doc_id)
    cache_key = f'llm:advisory:v3:{sha256(doc.text)}'   # same text -> same result
    if cached := cache.get(cache_key):
        return save_extract(doc_id, AdvisoryExtract.model_validate(cached))
    raw = llm.complete_json(
        system=EXTRACT_PROMPT_V3,                       # versioned prompt
        user=wrap_untrusted(doc.text[:50_000]),         # data, not instructions
        schema=AdvisoryExtract.model_json_schema(),
        timeout=60)
    try:
        result = AdvisoryExtract.model_validate(raw)
    except ValidationError:
        mark_for_human_review(doc_id, raw)               # is not accepted silently
        return
    cache.set(cache_key, result.model_dump(), timeout=30 * 86400)
    METRIC_LLM_TOKENS.labels(task='advisory').inc(raw.usage_total)
    save_extract(doc_id, result, model=llm.model_name, prompt_version=3)

# ======================================================================
# Q: Write an audit log that cannot easily be tampered with.
#
# The hash chain (each entry contains the hash of the previous one) shows
# whether someone modified or deleted an entry. For strict compliance, keep a
# copy in WORM storage (e.g. S3 Object Lock). Entries are written from a single
# place, in order, so the chain does not break.
# ======================================================================

class AuditLog(Base):
    __tablename__ = 'audit_log'
    id: Mapped[int] = mapped_column(BigInteger, primary_key=True)
    at: Mapped[datetime] = mapped_column(server_default=func.now())
    tenant_id: Mapped[int]
    actor_id: Mapped[int | None]
    action: Mapped[str]                       # 'reports:export'
    target: Mapped[str]                       # 'report:123'
    ip: Mapped[str | None]
    details: Mapped[dict] = mapped_column(JSONB, default=dict)
    prev_hash: Mapped[str]
    hash: Mapped[str]                         # sha256(prev_hash + entry)

-- In the database: the application can only write
REVOKE UPDATE, DELETE ON audit_log FROM app_user;

# ======================================================================
# Q: Find the security problems in this code.
#
# (1) SQL injection: an f-string inside text(). For values I use bind parameters
# (:x). For column names, which cannot be bound, I use an allowlist. (2) SSRF:
# the user can pass http://169.254.169.254/ and get AWS credentials from the
# metadata service. It needs a domain allowlist, blocking of private IP ranges
# after DNS resolution, a timeout and a size limit. There is also no timeout at
# all. (3) The token ends up in the logs. (4) There is no auth and no tenant
# filter: it returns all reports of all customers, and SELECT * may expose
# internal fields.
# ======================================================================

@app.get('/v1/reports')
def reports(sort: str, url: str, session=Depends(get_session)):
    rows = session.execute(text(f'SELECT * FROM reports ORDER BY {sort}'))   # (1)
    preview = requests.get(url).text                                         # (2)
    logger.info('user token %s', request.headers['Authorization'])           # (3)
    return {'rows': [dict(r._mapping) for r in rows], 'preview': preview}   # (4)

SORTABLE = {'created': Report.created_at, 'name': Report.name}

@app.get('/v1/reports', response_model=list[ReportOut])
def reports(sort: Literal['created', 'name'] = 'created',
            user: User = Depends(require_permission('reports:read')),
            session: Session = Depends(get_session)):
    stmt = (select(Report).where(Report.tenant_id == user.tenant_id)
            .order_by(SORTABLE[sort].desc()).limit(100))
    return session.scalars(stmt).all()

# ======================================================================
# Q: Which metrics do you measure? Write an example.
#
# For services, RED: Rate, Errors, Duration (p50/p95/p99, not the average). For
# resources, USE: Utilization, Saturation, Errors. And business metrics: imports
# per hour, rejected rows, queue length, delay from event to alert.
# Watch out for cardinality: never user_id or tenant_id as a label in Prometheus
# (millions of time series). Those go to logs/traces.
# ======================================================================

from prometheus_client import Counter, Histogram

IMPORT_ROWS = Counter('import_rows_total', 'Rows processed',
                      ['source', 'result'])          # result: ok|rejected
EXTERNAL_LATENCY = Histogram('external_api_seconds', 'External API latency',
                             ['provider'])

with EXTERNAL_LATENCY.labels(provider='geoip').time():
    data = client.lookup(ip)
IMPORT_ROWS.labels(source='edr', result='ok').inc(len(good))