Safe Haskell	None
Language	GHC2021

Ecluse.Core.Osv.Stream

Contents

Ingest bounds and drop accounting

Description

Streaming ingest of an osv.dev export archive, bounded against a pathological or tampered payload.

Pilot fetches <base>/<ecosystem>/all.zip from the public osv.dev mirror and decodes each advisory JSON on the way to compiling osv.db. osv.dev is trusted in normal operation, but the feed is an aggregation of many upstream databases: a single poisoned record can ride in with every transport header honest, so ingest is bounded as defence-in-depth. The bounds are generous (a real advisory is kilobytes) and fail-soft per entry: one bad advisory is dropped and logged, and the rest of the archive keeps flowing.

Two levels of response, tallied in IngestStats:

A single over-large or malformed entry is dropped and counted. The ilMaxAdvisoryBytes cap is enforced before the bytes are retained and before the JSON is decoded, so an inflation bomb never reaches the decoder whole; the offending entry is drained to its boundary so the following entries stay aligned.
An advisory that expands into more than ilMaxAdvisoryFanOut ranges is logged as anomalous but still ingested (log-only, non-gating).

The aggregate verdict is a separate, pure decision (systemicDrop): the compiler reads the tally once the stream completes and, if drops are systemic rather than isolated, abandons the run (PilotIngestAborted) so a consumer keeps its last-good artifact instead of adopting a hole-ridden one.

Depth is not guarded here on the decoded value: decodeStrict materialises the whole intermediate value before any post-decode check could run, and OsvAdvisory cannot represent unbounded nesting anyway, so the byte cap (which holds parse cost to a constant multiple of the input) plus the process heap ceiling resolved at boot (Ecluse.Rts) are what bound a small-but-deep payload.

Synopsis

streamOsvUrl :: forall (m :: Type -> Type) i. (MonadResource m, MonadThrow m, KatipContext m) => Maybe TracerProvider -> OsvIngest -> String -> ConduitT i ExtractedOsv m ()
parseOsvStream :: forall (m :: Type -> Type). (MonadResource m, MonadThrow m, KatipContext m) => Maybe TracerProvider -> OsvIngest -> ConduitT ByteString ExtractedOsv m ()
data IngestLimits = IngestLimits {
- ilMaxAdvisoryBytes :: !Int
- ilMaxAdvisoryFanOut :: !Int
}
defaultIngestLimits :: IngestLimits
data IngestStats = IngestStats {
- statAccepted :: !Int
- statDroppedOversize :: !Int
- statDroppedMalformed :: !Int
}
data IngestCounter
data OsvIngest = OsvIngest {
- ingestLimits :: IngestLimits
- ingestCounter :: IngestCounter
}
newOsvIngest :: MonadIO m => IngestLimits -> m OsvIngest
readIngestStats :: MonadIO m => OsvIngest -> m IngestStats
resetIngestStats :: MonadIO m => OsvIngest -> m ()
systemicDrop :: IngestStats -> Bool
newtype PilotIngestAborted = PilotIngestAborted IngestStats

Documentation

streamOsvUrl :: forall (m :: Type -> Type) i. (MonadResource m, MonadThrow m, KatipContext m) => Maybe TracerProvider -> OsvIngest -> String -> ConduitT i ExtractedOsv m () Source #

Fetch the OSV zip and stream its contents, bounded by ingest.

parseOsvStream :: forall (m :: Type -> Type). (MonadResource m, MonadThrow m, KatipContext m) => Maybe TracerProvider -> OsvIngest -> ConduitT ByteString ExtractedOsv m () Source #

Parse the zip stream and emit ExtractedOsv, bounded by ingest.

Ingest bounds and drop accounting

data IngestLimits Source #

The tunable per-advisory ingest bounds. Generous by design: osv.dev is trusted in normal operation, so these only backstop a pathological or tampered payload and must never trip on a real, if large, advisory.

Constructors

IngestLimits
Fields ilMaxAdvisoryBytes :: !Int Largest decompressed advisory JSON, in bytes, accepted from one zip entry before it is dropped. Bounds memory and, transitively, decode cost. ilMaxAdvisoryFanOut :: !Int Number of extracted ranges one advisory may expand into before it is flagged as anomalous. Log-only: the advisory is still ingested.

Instances

Instances details

Show IngestLimits Source #
Instance details Defined in Ecluse.Core.Osv.Stream Methods showsPrec :: Int -> IngestLimits -> ShowS # show :: IngestLimits -> String # showList :: [IngestLimits] -> ShowS #
Eq IngestLimits Source #
Instance details Defined in Ecluse.Core.Osv.Stream Methods (==) :: IngestLimits -> IngestLimits -> Bool # (/=) :: IngestLimits -> IngestLimits -> Bool #

defaultIngestLimits :: IngestLimits Source #

Sane defaults for IngestLimits: an 8 MiB per-advisory ceiling (a real advisory is kilobytes, so this is generous headroom) and a 256-range fan-out flag (a real advisory expands into a small multiple of its affected packages, far below this).

data IngestStats Source #

The running tally of one ingest pass: advisories accepted, and entries dropped by reason. Read once the stream completes to decide whether the artifact is trustworthy enough to publish (systemicDrop).

Constructors

IngestStats
Fields statAccepted :: !Int Advisory entries that decoded successfully. statDroppedOversize :: !Int Entries dropped for breaching `ilMaxAdvisoryBytes`. statDroppedMalformed :: !Int Entries dropped because their JSON did not decode.

Instances

Instances details

Show IngestStats Source #
Instance details Defined in Ecluse.Core.Osv.Stream Methods showsPrec :: Int -> IngestStats -> ShowS # show :: IngestStats -> String # showList :: [IngestStats] -> ShowS #
Eq IngestStats Source #
Instance details Defined in Ecluse.Core.Osv.Stream Methods (==) :: IngestStats -> IngestStats -> Bool # (/=) :: IngestStats -> IngestStats -> Bool #

data IngestCounter Source #

The mutable drop tally for one ingest pass. Opaque; read with readIngestStats.

data OsvIngest Source #

The context one ingest pass threads through the stream: its bounds and the live drop tally it records into.

Constructors

OsvIngest
Fields ingestLimits :: IngestLimits ingestCounter :: IngestCounter

newOsvIngest :: MonadIO m => IngestLimits -> m OsvIngest Source #

A fresh ingest context with the given bounds and a zeroed tally.

readIngestStats :: MonadIO m => OsvIngest -> m IngestStats Source #

Read the current drop tally.

resetIngestStats :: MonadIO m => OsvIngest -> m () Source #

Zero the tally. The compiler re-streams from a clean slate on each retry attempt, so the tally is reset alongside it and reflects only the final attempt.

systemicDrop :: IngestStats -> Bool Source #

Whether a run's drop tally signals systemic corruption (a hostile or broken feed) rather than a few poisoned records, so its artifact must not be published. Trips only when drops are both absolutely non-trivial and a large fraction of all entries, so a handful of bad advisories in a healthy feed never blocks a build, while a feed that is mostly unusable does.

newtype PilotIngestAborted Source #

Raised after a compile pass whose drop tally systemicDrop judged systemic: the run is abandoned without publishing, so a consumer keeps its last-good artifact rather than adopting a hole-ridden one.

Constructors

PilotIngestAborted IngestStats

Instances

Instances details

Exception PilotIngestAborted Source #
Instance details Defined in Ecluse.Core.Osv.Stream Methods toException :: PilotIngestAborted -> SomeException # fromException :: SomeException -> Maybe PilotIngestAborted # displayException :: PilotIngestAborted -> String # backtraceDesired :: PilotIngestAborted -> Bool #
Show PilotIngestAborted Source #
Instance details Defined in Ecluse.Core.Osv.Stream Methods showsPrec :: Int -> PilotIngestAborted -> ShowS # show :: PilotIngestAborted -> String # showList :: [PilotIngestAborted] -> ShowS #