stella-ops.org/git.stella-ops.org
1
0
Fork 0
Code Issues Pull Requests Actions 2 Releases Wiki Activity

save progress

Browse Source
This commit is contained in:
StellaOps Bot
2025-12-18 09:53:46 +02:00
parent 28823a8960
commit 7d5250238c
87 changed files with 9750 additions and 2026 deletions
Download Patch File Download Diff File Expand all files Collapse all files

33
src/Scanner/__Libraries/StellaOps.Scanner.CallGraph/Models/CallGraphModels.cs
View File

@@ -3,6 +3,7 @@ using System.Security.Cryptography;
using System.Text;
using System.Text.Json;
using System.Text.Json.Serialization;
using StellaOps.Scanner.CallGraph.Serialization;
using StellaOps.Scanner.Reachability;
namespace StellaOps.Scanner.CallGraph;
@@ -12,10 +13,18 @@ public sealed record CallGraphSnapshot(
[property: JsonPropertyName("graphDigest")] string GraphDigest,
[property: JsonPropertyName("language")] string Language,
[property: JsonPropertyName("extractedAt")] DateTimeOffset ExtractedAt,
[property: JsonPropertyName("nodes")] ImmutableArray<CallGraphNode> Nodes,
[property: JsonPropertyName("edges")] ImmutableArray<CallGraphEdge> Edges,
[property: JsonPropertyName("entrypointIds")] ImmutableArray<string> EntrypointIds,
[property: JsonPropertyName("sinkIds")] ImmutableArray<string> SinkIds)
[property: JsonPropertyName("nodes")]
[property: JsonConverter(typeof(ImmutableArrayJsonConverter<CallGraphNode>))]
ImmutableArray<CallGraphNode> Nodes,
[property: JsonPropertyName("edges")]
[property: JsonConverter(typeof(ImmutableArrayJsonConverter<CallGraphEdge>))]
ImmutableArray<CallGraphEdge> Edges,
[property: JsonPropertyName("entrypointIds")]
[property: JsonConverter(typeof(ImmutableArrayJsonConverter<string>))]
ImmutableArray<string> EntrypointIds,
[property: JsonPropertyName("sinkIds")]
[property: JsonConverter(typeof(ImmutableArrayJsonConverter<string>))]
ImmutableArray<string> SinkIds)
{
public CallGraphSnapshot Trimmed()
{
@@ -286,7 +295,9 @@ public static class CallGraphDigests
public sealed record ReachabilityPath(
[property: JsonPropertyName("entrypointId")] string EntrypointId,
[property: JsonPropertyName("sinkId")] string SinkId,
[property: JsonPropertyName("nodeIds")] ImmutableArray<string> NodeIds)
[property: JsonPropertyName("nodeIds")]
[property: JsonConverter(typeof(ImmutableArrayJsonConverter<string>))]
ImmutableArray<string> NodeIds)
{
public ReachabilityPath Trimmed()
{
@@ -309,9 +320,15 @@ public sealed record ReachabilityAnalysisResult(
[property: JsonPropertyName("graphDigest")] string GraphDigest,
[property: JsonPropertyName("language")] string Language,
[property: JsonPropertyName("computedAt")] DateTimeOffset ComputedAt,
[property: JsonPropertyName("reachableNodeIds")] ImmutableArray<string> ReachableNodeIds,
[property: JsonPropertyName("reachableSinkIds")] ImmutableArray<string> ReachableSinkIds,
[property: JsonPropertyName("paths")] ImmutableArray<ReachabilityPath> Paths,
[property: JsonPropertyName("reachableNodeIds")]
[property: JsonConverter(typeof(ImmutableArrayJsonConverter<string>))]
ImmutableArray<string> ReachableNodeIds,
[property: JsonPropertyName("reachableSinkIds")]
[property: JsonConverter(typeof(ImmutableArrayJsonConverter<string>))]
ImmutableArray<string> ReachableSinkIds,
[property: JsonPropertyName("paths")]
[property: JsonConverter(typeof(ImmutableArrayJsonConverter<ReachabilityPath>))]
ImmutableArray<ReachabilityPath> Paths,
[property: JsonPropertyName("resultDigest")] string ResultDigest)
{
public ReachabilityAnalysisResult Trimmed()

42
src/Scanner/__Libraries/StellaOps.Scanner.CallGraph/Serialization/ImmutableArrayJsonConverter.cs Normal file
View File

@@ -0,0 +1,42 @@
using System.Collections.Generic;
using System.Collections.Immutable;
using System.Text.Json;
using System.Text.Json.Serialization;
namespace StellaOps.Scanner.CallGraph.Serialization;
/// <summary>
/// System.Text.Json converter for <see cref="ImmutableArray{T}"/> to ensure default serializer options
/// can round-trip call graph models without requiring per-call JsonSerializerOptions registration.
/// </summary>
public sealed class ImmutableArrayJsonConverter<T> : JsonConverter<ImmutableArray<T>>
{
public override ImmutableArray<T> Read(ref Utf8JsonReader reader, Type typeToConvert, JsonSerializerOptions options)
{
if (reader.TokenType == JsonTokenType.Null)
{
return ImmutableArray<T>.Empty;
}
var values = JsonSerializer.Deserialize<List<T>>(ref reader, options);
if (values is null || values.Count == 0)
{
return ImmutableArray<T>.Empty;
}
return ImmutableArray.CreateRange(values);
}
public override void Write(Utf8JsonWriter writer, ImmutableArray<T> value, JsonSerializerOptions options)
{
writer.WriteStartArray();
var normalized = value.IsDefault ? ImmutableArray<T>.Empty : value;
foreach (var item in normalized)
{
JsonSerializer.Serialize(writer, item, options);
}
writer.WriteEndArray();
}
}

41
src/Scanner/__Libraries/StellaOps.Scanner.Storage/Epss/EpssBundleSource.cs Normal file
View File

@@ -0,0 +1,41 @@
// -----------------------------------------------------------------------------
// EpssBundleSource.cs
// Sprint: SPRINT_3410_0001_0001_epss_ingestion_storage
// Task: EPSS-3410-005
// Description: File-based EPSS source for air-gapped imports.
// -----------------------------------------------------------------------------
namespace StellaOps.Scanner.Storage.Epss;
public sealed class EpssBundleSource : IEpssSource
{
private readonly string _path;
public EpssBundleSource(string path)
{
ArgumentException.ThrowIfNullOrWhiteSpace(path);
_path = path;
}
public ValueTask<EpssSourceFile> GetAsync(DateOnly modelDate, CancellationToken cancellationToken = default)
{
cancellationToken.ThrowIfCancellationRequested();
var fileName = $"epss_scores-{modelDate:yyyy-MM-dd}.csv.gz";
var resolvedPath = _path;
if (Directory.Exists(_path))
{
resolvedPath = Path.Combine(_path, fileName);
}
if (!File.Exists(resolvedPath))
{
throw new FileNotFoundException($"EPSS bundle file not found: {resolvedPath}", resolvedPath);
}
var sourceUri = $"bundle://{Path.GetFileName(resolvedPath)}";
return ValueTask.FromResult(new EpssSourceFile(sourceUri, resolvedPath, deleteOnDispose: false));
}
}

75
src/Scanner/__Libraries/StellaOps.Scanner.Storage/Epss/EpssChangeDetector.cs Normal file
View File

@@ -0,0 +1,75 @@
// -----------------------------------------------------------------------------
// EpssChangeDetector.cs
// Sprint: SPRINT_3410_0001_0001_epss_ingestion_storage
// Task: EPSS-3410-008
// Description: Deterministic EPSS delta flag computation (mirrors SQL function).
// -----------------------------------------------------------------------------
namespace StellaOps.Scanner.Storage.Epss;
public static class EpssChangeDetector
{
public static EpssChangeThresholds DefaultThresholds => new(
HighScore: 0.50,
HighPercentile: 0.95,
BigJumpDelta: 0.10);
public static EpssChangeFlags ComputeFlags(
double? oldScore,
double newScore,
double? oldPercentile,
double newPercentile,
EpssChangeThresholds thresholds)
{
var flags = EpssChangeFlags.None;
if (oldScore is null)
{
flags |= EpssChangeFlags.NewScored;
}
if (oldScore is not null)
{
if (oldScore < thresholds.HighScore && newScore >= thresholds.HighScore)
{
flags |= EpssChangeFlags.CrossedHigh;
}
if (oldScore >= thresholds.HighScore && newScore < thresholds.HighScore)
{
flags |= EpssChangeFlags.CrossedLow;
}
var delta = newScore - oldScore.Value;
if (delta > thresholds.BigJumpDelta)
{
flags |= EpssChangeFlags.BigJumpUp;
}
if (delta < -thresholds.BigJumpDelta)
{
flags |= EpssChangeFlags.BigJumpDown;
}
}
if ((oldPercentile is null || oldPercentile < thresholds.HighPercentile)
&& newPercentile >= thresholds.HighPercentile)
{
flags |= EpssChangeFlags.TopPercentile;
}
if (oldPercentile is not null && oldPercentile >= thresholds.HighPercentile
&& newPercentile < thresholds.HighPercentile)
{
flags |= EpssChangeFlags.LeftTopPercentile;
}
return flags;
}
}
public readonly record struct EpssChangeThresholds(
double HighScore,
double HighPercentile,
double BigJumpDelta);

36
src/Scanner/__Libraries/StellaOps.Scanner.Storage/Epss/EpssChangeFlags.cs Normal file
View File

@@ -0,0 +1,36 @@
// -----------------------------------------------------------------------------
// EpssChangeFlags.cs
// Sprint: SPRINT_3410_0001_0001_epss_ingestion_storage
// Task: EPSS-3410-008
// Description: Flag bitmask for EPSS change detection.
// -----------------------------------------------------------------------------
namespace StellaOps.Scanner.Storage.Epss;
[Flags]
public enum EpssChangeFlags
{
None = 0,
/// <summary>0x01 - CVE newly scored (not in previous snapshot).</summary>
NewScored = 1,
/// <summary>0x02 - Crossed above the high score threshold.</summary>
CrossedHigh = 2,
/// <summary>0x04 - Crossed below the high score threshold.</summary>
CrossedLow = 4,
/// <summary>0x08 - Score increased by more than the big jump delta.</summary>
BigJumpUp = 8,
/// <summary>0x10 - Score decreased by more than the big jump delta.</summary>
BigJumpDown = 16,
/// <summary>0x20 - Entered the top percentile band.</summary>
TopPercentile = 32,
/// <summary>0x40 - Left the top percentile band.</summary>
LeftTopPercentile = 64
}

297
src/Scanner/__Libraries/StellaOps.Scanner.Storage/Epss/EpssCsvStreamParser.cs Normal file
View File

@@ -0,0 +1,297 @@
// -----------------------------------------------------------------------------
// EpssCsvStreamParser.cs
// Sprint: SPRINT_3410_0001_0001_epss_ingestion_storage
// Task: EPSS-3410-006
// Description: Streaming gzip CSV parser for EPSS snapshots with deterministic validation.
// -----------------------------------------------------------------------------
using System.IO.Compression;
using System.Runtime.CompilerServices;
using System.Security.Cryptography;
using System.Text;
using System.Text.RegularExpressions;
namespace StellaOps.Scanner.Storage.Epss;
public sealed class EpssCsvStreamParser
{
private static readonly Regex ModelVersionTagRegex = new(@"\bv\d{4}\.\d{2}\.\d{2}\b", RegexOptions.Compiled);
private static readonly Regex PublishedDateRegex = new(@"\b\d{4}-\d{2}-\d{2}\b", RegexOptions.Compiled);
public EpssCsvParseSession ParseGzip(Stream gzipStream)
=> new(gzipStream);
public sealed class EpssCsvParseSession : IAsyncEnumerable<EpssScoreRow>, IAsyncDisposable
{
private readonly Stream _gzipStream;
private bool _enumerated;
private bool _disposed;
public EpssCsvParseSession(Stream gzipStream)
{
_gzipStream = gzipStream ?? throw new ArgumentNullException(nameof(gzipStream));
}
public string? ModelVersionTag { get; private set; }
public DateOnly? PublishedDate { get; private set; }
public int RowCount { get; private set; }
public string? DecompressedSha256 { get; private set; }
public IAsyncEnumerator<EpssScoreRow> GetAsyncEnumerator(CancellationToken cancellationToken = default)
{
if (_disposed)
{
throw new ObjectDisposedException(nameof(EpssCsvParseSession));
}
if (_enumerated)
{
throw new InvalidOperationException("EPSS parse session can only be enumerated once.");
}
_enumerated = true;
return ParseAsync(cancellationToken).GetAsyncEnumerator(cancellationToken);
}
public ValueTask DisposeAsync()
{
if (_disposed)
{
return ValueTask.CompletedTask;
}
_disposed = true;
return _gzipStream.DisposeAsync();
}
private async IAsyncEnumerable<EpssScoreRow> ParseAsync([EnumeratorCancellation] CancellationToken cancellationToken)
{
await using var gzip = new GZipStream(_gzipStream, CompressionMode.Decompress, leaveOpen: false);
await using var hashing = new HashingReadStream(gzip);
using var reader = new StreamReader(
hashing,
Encoding.UTF8,
detectEncodingFromByteOrderMarks: true,
bufferSize: 64 * 1024,
leaveOpen: true);
string? line;
while ((line = await reader.ReadLineAsync(cancellationToken).ConfigureAwait(false)) is not null)
{
cancellationToken.ThrowIfCancellationRequested();
if (line.StartsWith('#'))
{
ParseCommentLine(line);
continue;
}
// First non-comment line is the CSV header.
var header = line.Trim();
if (!header.Equals("cve,epss,percentile", StringComparison.OrdinalIgnoreCase))
{
throw new FormatException($"Unexpected EPSS CSV header: '{header}'. Expected 'cve,epss,percentile'.");
}
break;
}
if (line is null)
{
throw new FormatException("EPSS CSV appears to be empty.");
}
while ((line = await reader.ReadLineAsync(cancellationToken).ConfigureAwait(false)) is not null)
{
cancellationToken.ThrowIfCancellationRequested();
if (string.IsNullOrWhiteSpace(line))
{
continue;
}
var row = ParseRow(line);
RowCount++;
yield return row;
}
DecompressedSha256 = "sha256:" + hashing.GetHashHex();
}
private void ParseCommentLine(string line)
{
if (ModelVersionTag is null)
{
var match = ModelVersionTagRegex.Match(line);
if (match.Success)
{
ModelVersionTag = match.Value;
}
}
if (PublishedDate is null)
{
var match = PublishedDateRegex.Match(line);
if (match.Success && DateOnly.TryParseExact(match.Value, "yyyy-MM-dd", out var date))
{
PublishedDate = date;
}
}
}
private static EpssScoreRow ParseRow(string line)
{
var comma1 = line.IndexOf(',');
if (comma1 <= 0)
{
throw new FormatException($"Invalid EPSS CSV row: '{line}'.");
}
var comma2 = line.IndexOf(',', comma1 + 1);
if (comma2 <= comma1 + 1 || comma2 == line.Length - 1)
{
throw new FormatException($"Invalid EPSS CSV row: '{line}'.");
}
var cveSpan = line.AsSpan(0, comma1).Trim();
var scoreSpan = line.AsSpan(comma1 + 1, comma2 - comma1 - 1).Trim();
var percentileSpan = line.AsSpan(comma2 + 1).Trim();
var cveId = NormalizeCveId(cveSpan);
if (!double.TryParse(scoreSpan, System.Globalization.NumberStyles.Float, System.Globalization.CultureInfo.InvariantCulture, out var score))
{
throw new FormatException($"Invalid EPSS score value in row: '{line}'.");
}
if (!double.TryParse(percentileSpan, System.Globalization.NumberStyles.Float, System.Globalization.CultureInfo.InvariantCulture, out var percentile))
{
throw new FormatException($"Invalid EPSS percentile value in row: '{line}'.");
}
if (score < 0.0 || score > 1.0)
{
throw new FormatException($"EPSS score out of range [0,1] in row: '{line}'.");
}
if (percentile < 0.0 || percentile > 1.0)
{
throw new FormatException($"EPSS percentile out of range [0,1] in row: '{line}'.");
}
return new EpssScoreRow(cveId, score, percentile);
}
private static string NormalizeCveId(ReadOnlySpan<char> value)
{
if (value.Length == 0)
{
throw new FormatException("EPSS row has empty CVE ID.");
}
// Expected: CVE-YYYY-NNNN...
if (value.Length < "CVE-1999-0000".Length)
{
throw new FormatException($"Invalid CVE ID '{value.ToString()}'.");
}
if (!value.StartsWith("CVE-", StringComparison.OrdinalIgnoreCase))
{
throw new FormatException($"Invalid CVE ID '{value.ToString()}'.");
}
var normalized = value.ToString().ToUpperInvariant();
return normalized;
}
}
private sealed class HashingReadStream : Stream
{
private readonly Stream _inner;
private readonly IncrementalHash _hash = IncrementalHash.CreateHash(HashAlgorithmName.SHA256);
private bool _disposed;
private string? _sha256Hex;
public HashingReadStream(Stream inner)
{
_inner = inner ?? throw new ArgumentNullException(nameof(inner));
}
public string GetHashHex()
{
if (_sha256Hex is not null)
{
return _sha256Hex;
}
var digest = _hash.GetHashAndReset();
_sha256Hex = Convert.ToHexString(digest).ToLowerInvariant();
return _sha256Hex;
}
public override bool CanRead => !_disposed && _inner.CanRead;
public override bool CanSeek => false;
public override bool CanWrite => false;
public override long Length => throw new NotSupportedException();
public override long Position { get => throw new NotSupportedException(); set => throw new NotSupportedException(); }
public override void Flush() => throw new NotSupportedException();
public override long Seek(long offset, SeekOrigin origin) => throw new NotSupportedException();
public override void SetLength(long value) => throw new NotSupportedException();
public override void Write(byte[] buffer, int offset, int count) => throw new NotSupportedException();
public override int Read(byte[] buffer, int offset, int count)
{
var read = _inner.Read(buffer, offset, count);
if (read > 0)
{
_hash.AppendData(buffer, offset, read);
}
return read;
}
public override async ValueTask<int> ReadAsync(Memory<byte> buffer, CancellationToken cancellationToken = default)
{
var read = await _inner.ReadAsync(buffer, cancellationToken).ConfigureAwait(false);
if (read > 0)
{
var slice = buffer.Slice(0, read);
_hash.AppendData(slice.Span);
}
return read;
}
protected override void Dispose(bool disposing)
{
if (_disposed)
{
return;
}
if (disposing)
{
_hash.Dispose();
_inner.Dispose();
}
_disposed = true;
base.Dispose(disposing);
}
public override async ValueTask DisposeAsync()
{
if (_disposed)
{
return;
}
_hash.Dispose();
await _inner.DisposeAsync().ConfigureAwait(false);
_disposed = true;
await base.DisposeAsync().ConfigureAwait(false);
}
}
}

46
src/Scanner/__Libraries/StellaOps.Scanner.Storage/Epss/EpssOnlineSource.cs Normal file
View File

@@ -0,0 +1,46 @@
// -----------------------------------------------------------------------------
// EpssOnlineSource.cs
// Sprint: SPRINT_3410_0001_0001_epss_ingestion_storage
// Task: EPSS-3410-004
// Description: Online EPSS source that downloads FIRST.org CSV.gz snapshots.
// -----------------------------------------------------------------------------
using System.Net.Http;
namespace StellaOps.Scanner.Storage.Epss;
public sealed class EpssOnlineSource : IEpssSource
{
public const string DefaultBaseUri = "https://epss.empiricalsecurity.com/";
private readonly HttpClient _httpClient;
private readonly Uri _baseUri;
public EpssOnlineSource(HttpClient httpClient, string? baseUri = null)
{
_httpClient = httpClient ?? throw new ArgumentNullException(nameof(httpClient));
_baseUri = new Uri(string.IsNullOrWhiteSpace(baseUri) ? DefaultBaseUri : baseUri, UriKind.Absolute);
}
public async ValueTask<EpssSourceFile> GetAsync(DateOnly modelDate, CancellationToken cancellationToken = default)
{
var fileName = $"epss_scores-{modelDate:yyyy-MM-dd}.csv.gz";
var uri = new Uri(_baseUri, fileName);
var tempPath = Path.Combine(
Path.GetTempPath(),
$"stellaops-epss-{Guid.NewGuid():n}-{fileName}");
using var response = await _httpClient.GetAsync(uri, HttpCompletionOption.ResponseHeadersRead, cancellationToken).ConfigureAwait(false);
response.EnsureSuccessStatusCode();
await using var sourceStream = await response.Content.ReadAsStreamAsync(cancellationToken).ConfigureAwait(false);
await using (var destinationStream = new FileStream(tempPath, FileMode.CreateNew, FileAccess.Write, FileShare.None))
{
await sourceStream.CopyToAsync(destinationStream, cancellationToken).ConfigureAwait(false);
}
return new EpssSourceFile(uri.ToString(), tempPath, deleteOnDispose: true);
}
}

17
src/Scanner/__Libraries/StellaOps.Scanner.Storage/Epss/EpssScoreRow.cs Normal file
View File

@@ -0,0 +1,17 @@
// -----------------------------------------------------------------------------
// EpssScoreRow.cs
// Sprint: SPRINT_3410_0001_0001_epss_ingestion_storage
// Task: EPSS-3410-002
// Description: DTO representing a parsed EPSS CSV row.
// -----------------------------------------------------------------------------
namespace StellaOps.Scanner.Storage.Epss;
/// <summary>
/// Represents a single row from an EPSS CSV snapshot.
/// </summary>
public readonly record struct EpssScoreRow(
string CveId,
double Score,
double Percentile);

46
src/Scanner/__Libraries/StellaOps.Scanner.Storage/Epss/EpssSourceFile.cs Normal file
View File

@@ -0,0 +1,46 @@
// -----------------------------------------------------------------------------
// EpssSourceFile.cs
// Sprint: SPRINT_3410_0001_0001_epss_ingestion_storage
// Task: EPSS-3410-003
// Description: Local file materialization wrapper for EPSS sources.
// -----------------------------------------------------------------------------
namespace StellaOps.Scanner.Storage.Epss;
public sealed class EpssSourceFile : IAsyncDisposable
{
public EpssSourceFile(string sourceUri, string localPath, bool deleteOnDispose)
{
ArgumentException.ThrowIfNullOrWhiteSpace(sourceUri);
ArgumentException.ThrowIfNullOrWhiteSpace(localPath);
SourceUri = sourceUri;
LocalPath = localPath;
DeleteOnDispose = deleteOnDispose;
}
public string SourceUri { get; }
public string LocalPath { get; }
public bool DeleteOnDispose { get; }
public ValueTask DisposeAsync()
{
if (DeleteOnDispose)
{
try
{
if (File.Exists(LocalPath))
{
File.Delete(LocalPath);
}
}
catch
{
// Best-effort cleanup only.
}
}
return ValueTask.CompletedTask;
}
}

14
src/Scanner/__Libraries/StellaOps.Scanner.Storage/Epss/IEpssSource.cs Normal file
View File

@@ -0,0 +1,14 @@
// -----------------------------------------------------------------------------
// IEpssSource.cs
// Sprint: SPRINT_3410_0001_0001_epss_ingestion_storage
// Task: EPSS-3410-003
// Description: Abstraction for online vs air-gapped EPSS sources.
// -----------------------------------------------------------------------------
namespace StellaOps.Scanner.Storage.Epss;
public interface IEpssSource
{
ValueTask<EpssSourceFile> GetAsync(DateOnly modelDate, CancellationToken cancellationToken = default);
}

3
src/Scanner/__Libraries/StellaOps.Scanner.Storage/Extensions/ServiceCollectionExtensions.cs
View File

@@ -16,6 +16,7 @@ using StellaOps.Scanner.Storage.ObjectStore;
using StellaOps.Scanner.Storage.Postgres;
using StellaOps.Scanner.Storage.Repositories;
using StellaOps.Scanner.Storage.Services;
using StellaOps.Scanner.Storage.Epss;
namespace StellaOps.Scanner.Storage.Extensions;
@@ -81,6 +82,8 @@ public static class ServiceCollectionExtensions
services.AddScoped<IReachabilityResultRepository, PostgresReachabilityResultRepository>();
services.AddScoped<ICodeChangeRepository, PostgresCodeChangeRepository>();
services.AddScoped<IReachabilityDriftResultRepository, PostgresReachabilityDriftResultRepository>();
services.AddSingleton<EpssCsvStreamParser>();
services.AddScoped<IEpssRepository, PostgresEpssRepository>();
services.AddSingleton<IEntryTraceResultStore, EntryTraceResultStore>();
services.AddSingleton<IRubyPackageInventoryStore, RubyPackageInventoryStore>();
services.AddSingleton<IBunPackageInventoryStore, BunPackageInventoryStore>();

78
src/Scanner/__Libraries/StellaOps.Scanner.Storage/Postgres/Migrations/011_epss_raw_layer.sql Normal file
View File

@@ -0,0 +1,78 @@
-- SPDX-License-Identifier: AGPL-3.0-or-later
-- Sprint: 3413
-- Task: EPSS Raw Feed Layer
-- Description: Creates epss_raw table for immutable full payload storage
-- Enables deterministic replay without re-downloading from FIRST.org
-- Advisory: 18-Dec-2025 - Designing a Layered EPSS v4 Database.md
-- ============================================================================
-- EPSS Raw Feed Storage (Immutable)
-- ============================================================================
-- Layer 1 of 3-layer EPSS architecture
-- Stores full CSV payload as JSONB for deterministic replay capability
-- Expected storage: ~15MB/day compressed → ~5GB/year in JSONB
CREATE TABLE IF NOT EXISTS epss_raw (
raw_id BIGSERIAL PRIMARY KEY,
source_uri TEXT NOT NULL,
asof_date DATE NOT NULL,
ingestion_ts TIMESTAMPTZ NOT NULL DEFAULT now(),
-- Full payload storage
payload JSONB NOT NULL, -- Full CSV content as JSON array of {cve, epss, percentile}
payload_sha256 BYTEA NOT NULL, -- SHA-256 of decompressed content for integrity
-- Metadata extracted from CSV comment line
header_comment TEXT, -- Leading # comment if present (e.g., "# model: v2025.03.14...")
model_version TEXT, -- Extracted model version (e.g., "v2025.03.14")
published_date DATE, -- Extracted publish date from comment
-- Stats
row_count INT NOT NULL,
compressed_size BIGINT, -- Original .csv.gz file size
decompressed_size BIGINT, -- Decompressed CSV size
-- Link to import run (optional, for correlation)
import_run_id UUID REFERENCES epss_import_runs(import_run_id),
-- Idempotency: same source + date + content hash = same record
CONSTRAINT epss_raw_unique UNIQUE (source_uri, asof_date, payload_sha256)
);
-- Performance indexes
CREATE INDEX IF NOT EXISTS idx_epss_raw_asof
ON epss_raw (asof_date DESC);
CREATE INDEX IF NOT EXISTS idx_epss_raw_model
ON epss_raw (model_version);
CREATE INDEX IF NOT EXISTS idx_epss_raw_import_run
ON epss_raw (import_run_id);
-- Comments
COMMENT ON TABLE epss_raw IS 'Layer 1: Immutable raw EPSS payload storage for deterministic replay';
COMMENT ON COLUMN epss_raw.payload IS 'Full CSV content as JSON array: [{cve:"CVE-...", epss:0.123, percentile:0.456}, ...]';
COMMENT ON COLUMN epss_raw.payload_sha256 IS 'SHA-256 hash of decompressed CSV for integrity verification';
COMMENT ON COLUMN epss_raw.header_comment IS 'Raw comment line from CSV (e.g., "# model: v2025.03.14, published: 2025-03-14")';
COMMENT ON COLUMN epss_raw.model_version IS 'Extracted model version for detecting model changes';
-- ============================================================================
-- Retention Policy Helper
-- ============================================================================
-- Function to prune old raw data (default: keep 365 days)
CREATE OR REPLACE FUNCTION prune_epss_raw(retention_days INT DEFAULT 365)
RETURNS INT AS $$
DECLARE
deleted_count INT;
BEGIN
DELETE FROM epss_raw
WHERE asof_date < CURRENT_DATE - retention_days::INTERVAL;
GET DIAGNOSTICS deleted_count = ROW_COUNT;
RAISE NOTICE 'Pruned % epss_raw records older than % days', deleted_count, retention_days;
RETURN deleted_count;
END;
$$ LANGUAGE plpgsql;
COMMENT ON FUNCTION prune_epss_raw IS 'Prunes epss_raw records older than retention_days (default: 365)';

179
src/Scanner/__Libraries/StellaOps.Scanner.Storage/Postgres/Migrations/012_epss_signal_layer.sql Normal file
View File

@@ -0,0 +1,179 @@
-- SPDX-License-Identifier: AGPL-3.0-or-later
-- Sprint: 3413
-- Task: EPSS Signal-Ready Layer
-- Description: Creates epss_signal table for tenant-scoped actionable events
-- Reduces noise by only signaling for observed CVEs per tenant
-- Advisory: 18-Dec-2025 - Designing a Layered EPSS v4 Database.md
-- ============================================================================
-- EPSS Signal-Ready Events (Tenant-Scoped)
-- ============================================================================
-- Layer 3 of 3-layer EPSS architecture
-- Pre-computed actionable events scoped to observed CVEs per tenant
-- Supports deduplication via dedupe_key and audit trail via explain_hash
CREATE TABLE IF NOT EXISTS epss_signal (
signal_id BIGSERIAL PRIMARY KEY,
tenant_id UUID NOT NULL,
model_date DATE NOT NULL,
cve_id TEXT NOT NULL,
-- Event classification
event_type TEXT NOT NULL, -- 'RISK_SPIKE', 'BAND_CHANGE', 'NEW_HIGH', 'MODEL_UPDATED'
risk_band TEXT, -- 'CRITICAL', 'HIGH', 'MEDIUM', 'LOW'
-- EPSS metrics at signal time
epss_score DOUBLE PRECISION,
epss_delta DOUBLE PRECISION, -- Delta from previous day
percentile DOUBLE PRECISION,
percentile_delta DOUBLE PRECISION, -- Delta from previous day
-- Model version tracking
is_model_change BOOLEAN NOT NULL DEFAULT false, -- True when FIRST.org updated model version
model_version TEXT,
-- Idempotency and audit
dedupe_key TEXT NOT NULL, -- Deterministic key for deduplication
explain_hash BYTEA NOT NULL, -- SHA-256 of signal inputs for audit trail
payload JSONB NOT NULL, -- Full evidence payload for downstream consumers
-- Timestamps
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
-- Deduplication constraint: same tenant + dedupe_key = same signal
CONSTRAINT epss_signal_dedupe UNIQUE (tenant_id, dedupe_key)
);
-- Performance indexes
CREATE INDEX IF NOT EXISTS idx_epss_signal_tenant_date
ON epss_signal (tenant_id, model_date DESC);
CREATE INDEX IF NOT EXISTS idx_epss_signal_tenant_cve
ON epss_signal (tenant_id, cve_id, model_date DESC);
CREATE INDEX IF NOT EXISTS idx_epss_signal_event_type
ON epss_signal (tenant_id, event_type, model_date DESC);
CREATE INDEX IF NOT EXISTS idx_epss_signal_risk_band
ON epss_signal (tenant_id, risk_band, model_date DESC)
WHERE risk_band IN ('CRITICAL', 'HIGH');
CREATE INDEX IF NOT EXISTS idx_epss_signal_model_change
ON epss_signal (model_date)
WHERE is_model_change = true;
-- Comments
COMMENT ON TABLE epss_signal IS 'Layer 3: Tenant-scoped EPSS signal events for actionable notifications';
COMMENT ON COLUMN epss_signal.event_type IS 'Event classification: RISK_SPIKE (delta > threshold), BAND_CHANGE (band transition), NEW_HIGH (new CVE in high percentile), MODEL_UPDATED (FIRST.org model version change)';
COMMENT ON COLUMN epss_signal.risk_band IS 'Derived risk band: CRITICAL (>=99.5%), HIGH (>=99%), MEDIUM (>=90%), LOW (<90%)';
COMMENT ON COLUMN epss_signal.is_model_change IS 'True when FIRST.org updated model version (v3->v4 etc), used to suppress noisy delta signals';
COMMENT ON COLUMN epss_signal.dedupe_key IS 'Deterministic key: {model_date}:{cve_id}:{event_type}:{band_before}->{band_after}';
COMMENT ON COLUMN epss_signal.explain_hash IS 'SHA-256 of signal inputs for deterministic audit trail';
COMMENT ON COLUMN epss_signal.payload IS 'Full evidence: {source, metrics, decision, thresholds, evidence_refs}';
-- ============================================================================
-- Signal Event Types Enum (for reference)
-- ============================================================================
-- Not enforced as constraint to allow future extensibility
-- Event Types:
-- - RISK_SPIKE: EPSS delta exceeds big_jump_delta threshold (default: 0.10)
-- - BAND_CHANGE: Risk band transition (e.g., MEDIUM -> HIGH)
-- - NEW_HIGH: CVE newly scored in high percentile (>=95th)
-- - DROPPED_LOW: CVE dropped below low percentile threshold
-- - MODEL_UPDATED: Summary event when FIRST.org updates model version
-- ============================================================================
-- Risk Band Configuration (per tenant)
-- ============================================================================
CREATE TABLE IF NOT EXISTS epss_signal_config (
config_id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
tenant_id UUID NOT NULL,
-- Thresholds for risk banding
critical_percentile DOUBLE PRECISION NOT NULL DEFAULT 0.995, -- Top 0.5%
high_percentile DOUBLE PRECISION NOT NULL DEFAULT 0.99, -- Top 1%
medium_percentile DOUBLE PRECISION NOT NULL DEFAULT 0.90, -- Top 10%
-- Thresholds for signal generation
big_jump_delta DOUBLE PRECISION NOT NULL DEFAULT 0.10, -- 10 percentage points
suppress_on_model_change BOOLEAN NOT NULL DEFAULT true, -- Suppress RISK_SPIKE on model change
-- Notification preferences
enabled_event_types TEXT[] NOT NULL DEFAULT ARRAY['RISK_SPIKE', 'BAND_CHANGE', 'NEW_HIGH'],
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
CONSTRAINT epss_signal_config_tenant_unique UNIQUE (tenant_id)
);
-- Comments
COMMENT ON TABLE epss_signal_config IS 'Per-tenant configuration for EPSS signal generation';
COMMENT ON COLUMN epss_signal_config.suppress_on_model_change IS 'When true, suppress RISK_SPIKE and BAND_CHANGE signals on model version change days';
-- ============================================================================
-- Helper Functions
-- ============================================================================
-- Compute risk band from percentile
CREATE OR REPLACE FUNCTION compute_epss_risk_band(
p_percentile DOUBLE PRECISION,
p_critical_threshold DOUBLE PRECISION DEFAULT 0.995,
p_high_threshold DOUBLE PRECISION DEFAULT 0.99,
p_medium_threshold DOUBLE PRECISION DEFAULT 0.90
) RETURNS TEXT AS $$
BEGIN
IF p_percentile >= p_critical_threshold THEN
RETURN 'CRITICAL';
ELSIF p_percentile >= p_high_threshold THEN
RETURN 'HIGH';
ELSIF p_percentile >= p_medium_threshold THEN
RETURN 'MEDIUM';
ELSE
RETURN 'LOW';
END IF;
END;
$$ LANGUAGE plpgsql IMMUTABLE;
COMMENT ON FUNCTION compute_epss_risk_band IS 'Computes risk band from percentile using configurable thresholds';
-- Compute dedupe key for signal
CREATE OR REPLACE FUNCTION compute_epss_signal_dedupe_key(
p_model_date DATE,
p_cve_id TEXT,
p_event_type TEXT,
p_old_band TEXT,
p_new_band TEXT
) RETURNS TEXT AS $$
BEGIN
RETURN format('%s:%s:%s:%s->%s',
p_model_date::TEXT,
p_cve_id,
p_event_type,
COALESCE(p_old_band, 'NONE'),
COALESCE(p_new_band, 'NONE')
);
END;
$$ LANGUAGE plpgsql IMMUTABLE;
COMMENT ON FUNCTION compute_epss_signal_dedupe_key IS 'Computes deterministic deduplication key for EPSS signals';
-- ============================================================================
-- Retention Policy Helper
-- ============================================================================
CREATE OR REPLACE FUNCTION prune_epss_signals(retention_days INT DEFAULT 90)
RETURNS INT AS $$
DECLARE
deleted_count INT;
BEGIN
DELETE FROM epss_signal
WHERE model_date < CURRENT_DATE - retention_days::INTERVAL;
GET DIAGNOSTICS deleted_count = ROW_COUNT;
RAISE NOTICE 'Pruned % epss_signal records older than % days', deleted_count, retention_days;
RETURN deleted_count;
END;
$$ LANGUAGE plpgsql;
COMMENT ON FUNCTION prune_epss_signals IS 'Prunes epss_signal records older than retention_days (default: 90)';

601
src/Scanner/__Libraries/StellaOps.Scanner.Storage/Postgres/PostgresEpssRepository.cs Normal file
View File

@@ -0,0 +1,601 @@
// -----------------------------------------------------------------------------
// PostgresEpssRepository.cs
// Sprint: SPRINT_3410_0001_0001_epss_ingestion_storage
// Tasks: EPSS-3410-007, EPSS-3410-008
// Description: PostgreSQL persistence for EPSS import runs, time-series scores, current projection, and change log.
// -----------------------------------------------------------------------------
using System.Data;
using Dapper;
using Npgsql;
using NpgsqlTypes;
using StellaOps.Scanner.Storage.Epss;
using StellaOps.Scanner.Storage.Repositories;
namespace StellaOps.Scanner.Storage.Postgres;
public sealed class PostgresEpssRepository : IEpssRepository
{
private static int _typeHandlersRegistered;
private readonly ScannerDataSource _dataSource;
private string SchemaName => _dataSource.SchemaName ?? ScannerDataSource.DefaultSchema;
private string ImportRunsTable => $"{SchemaName}.epss_import_runs";
private string ScoresTable => $"{SchemaName}.epss_scores";
private string CurrentTable => $"{SchemaName}.epss_current";
private string ChangesTable => $"{SchemaName}.epss_changes";
private string ConfigTable => $"{SchemaName}.epss_config";
public PostgresEpssRepository(ScannerDataSource dataSource)
{
EnsureTypeHandlers();
_dataSource = dataSource ?? throw new ArgumentNullException(nameof(dataSource));
}
public async Task<EpssImportRun?> GetImportRunAsync(DateOnly modelDate, CancellationToken cancellationToken = default)
{
var sql = $"""
SELECT
import_run_id,
model_date,
source_uri,
retrieved_at,
file_sha256,
decompressed_sha256,
row_count,
model_version_tag,
published_date,
status,
error,
created_at
FROM {ImportRunsTable}
WHERE model_date = @ModelDate
""";
await using var connection = await _dataSource.OpenSystemConnectionAsync(cancellationToken).ConfigureAwait(false);
var row = await connection.QuerySingleOrDefaultAsync<ImportRunRow>(
new CommandDefinition(sql, new { ModelDate = modelDate }, cancellationToken: cancellationToken)).ConfigureAwait(false);
return row?.ToModel();
}
public async Task<EpssImportRun> BeginImportAsync(
DateOnly modelDate,
string sourceUri,
DateTimeOffset retrievedAtUtc,
string fileSha256,
CancellationToken cancellationToken = default)
{
ArgumentException.ThrowIfNullOrWhiteSpace(sourceUri);
ArgumentException.ThrowIfNullOrWhiteSpace(fileSha256);
var insertSql = $"""
INSERT INTO {ImportRunsTable} (
model_date,
source_uri,
retrieved_at,
file_sha256,
row_count,
status,
created_at
) VALUES (
@ModelDate,
@SourceUri,
@RetrievedAtUtc,
@FileSha256,
0,
'PENDING',
@RetrievedAtUtc
)
ON CONFLICT (model_date) DO UPDATE SET
source_uri = EXCLUDED.source_uri,
retrieved_at = EXCLUDED.retrieved_at,
file_sha256 = EXCLUDED.file_sha256,
decompressed_sha256 = NULL,
row_count = 0,
model_version_tag = NULL,
published_date = NULL,
status = 'PENDING',
error = NULL
WHERE {ImportRunsTable}.status <> 'SUCCEEDED'
RETURNING
import_run_id,
model_date,
source_uri,
retrieved_at,
file_sha256,
decompressed_sha256,
row_count,
model_version_tag,
published_date,
status,
error,
created_at
""";
await using var connection = await _dataSource.OpenSystemConnectionAsync(cancellationToken).ConfigureAwait(false);
var row = await connection.QuerySingleOrDefaultAsync<ImportRunRow>(new CommandDefinition(
insertSql,
new
{
ModelDate = modelDate,
SourceUri = sourceUri,
RetrievedAtUtc = retrievedAtUtc,
FileSha256 = fileSha256
},
cancellationToken: cancellationToken)).ConfigureAwait(false);
if (row is not null)
{
return row.ToModel();
}
// Existing SUCCEEDED run: return it to allow the caller to decide idempotent behavior.
var existing = await GetImportRunAsync(modelDate, cancellationToken).ConfigureAwait(false);
if (existing is null)
{
throw new InvalidOperationException("EPSS import run conflict detected but existing row was not found.");
}
return existing;
}
public async Task MarkImportSucceededAsync(
Guid importRunId,
int rowCount,
string? decompressedSha256,
string? modelVersionTag,
DateOnly? publishedDate,
CancellationToken cancellationToken = default)
{
var sql = $"""
UPDATE {ImportRunsTable}
SET status = 'SUCCEEDED',
error = NULL,
row_count = @RowCount,
decompressed_sha256 = @DecompressedSha256,
model_version_tag = @ModelVersionTag,
published_date = @PublishedDate
WHERE import_run_id = @ImportRunId
""";
await using var connection = await _dataSource.OpenSystemConnectionAsync(cancellationToken).ConfigureAwait(false);
await connection.ExecuteAsync(new CommandDefinition(
sql,
new
{
ImportRunId = importRunId,
RowCount = rowCount,
DecompressedSha256 = decompressedSha256,
ModelVersionTag = modelVersionTag,
PublishedDate = publishedDate
},
cancellationToken: cancellationToken)).ConfigureAwait(false);
}
public async Task MarkImportFailedAsync(Guid importRunId, string error, CancellationToken cancellationToken = default)
{
ArgumentException.ThrowIfNullOrWhiteSpace(error);
var sql = $"""
UPDATE {ImportRunsTable}
SET status = 'FAILED',
error = @Error
WHERE import_run_id = @ImportRunId
""";
await using var connection = await _dataSource.OpenSystemConnectionAsync(cancellationToken).ConfigureAwait(false);
await connection.ExecuteAsync(new CommandDefinition(
sql,
new { ImportRunId = importRunId, Error = error },
cancellationToken: cancellationToken)).ConfigureAwait(false);
}
public async Task<EpssWriteResult> WriteSnapshotAsync(
Guid importRunId,
DateOnly modelDate,
DateTimeOffset updatedAtUtc,
IAsyncEnumerable<EpssScoreRow> rows,
CancellationToken cancellationToken = default)
{
ArgumentNullException.ThrowIfNull(rows);
await using var connection = await _dataSource.OpenSystemConnectionAsync(cancellationToken).ConfigureAwait(false);
await using var transaction = await connection.BeginTransactionAsync(cancellationToken).ConfigureAwait(false);
try
{
await EnsurePartitionsAsync(connection, transaction, modelDate, cancellationToken).ConfigureAwait(false);
const string stageTable = "epss_stage";
var createStageSql = $"""
CREATE TEMP TABLE {stageTable} (
cve_id TEXT NOT NULL,
epss_score DOUBLE PRECISION NOT NULL,
percentile DOUBLE PRECISION NOT NULL
) ON COMMIT DROP
""";
await connection.ExecuteAsync(new CommandDefinition(
createStageSql,
transaction: transaction,
cancellationToken: cancellationToken)).ConfigureAwait(false);
var (rowCount, distinctCount) = await CopyStageAsync(connection, transaction, stageTable, rows, cancellationToken).ConfigureAwait(false);
if (rowCount != distinctCount)
{
throw new InvalidOperationException($"EPSS staging table contains duplicate CVE IDs (rows={rowCount}, distinct={distinctCount}).");
}
var insertScoresSql = $"""
INSERT INTO {ScoresTable} (model_date, cve_id, epss_score, percentile, import_run_id)
SELECT @ModelDate, cve_id, epss_score, percentile, @ImportRunId
FROM {stageTable}
""";
await connection.ExecuteAsync(new CommandDefinition(
insertScoresSql,
new { ModelDate = modelDate, ImportRunId = importRunId },
transaction: transaction,
cancellationToken: cancellationToken)).ConfigureAwait(false);
await InsertChangesAsync(connection, transaction, stageTable, modelDate, importRunId, cancellationToken).ConfigureAwait(false);
await UpsertCurrentAsync(connection, transaction, stageTable, modelDate, importRunId, updatedAtUtc, cancellationToken).ConfigureAwait(false);
await transaction.CommitAsync(cancellationToken).ConfigureAwait(false);
return new EpssWriteResult(RowCount: rowCount, DistinctCveCount: distinctCount);
}
catch
{
await transaction.RollbackAsync(cancellationToken).ConfigureAwait(false);
throw;
}
}
public async Task<IReadOnlyDictionary<string, EpssCurrentEntry>> GetCurrentAsync(
IEnumerable<string> cveIds,
CancellationToken cancellationToken = default)
{
if (cveIds is null)
{
return new Dictionary<string, EpssCurrentEntry>(StringComparer.Ordinal);
}
var normalized = cveIds
.Where(static id => !string.IsNullOrWhiteSpace(id))
.Select(static id => id.Trim().ToUpperInvariant())
.Distinct(StringComparer.Ordinal)
.OrderBy(static id => id, StringComparer.Ordinal)
.ToArray();
if (normalized.Length == 0)
{
return new Dictionary<string, EpssCurrentEntry>(StringComparer.Ordinal);
}
var sql = $"""
SELECT cve_id, epss_score, percentile, model_date, import_run_id
FROM {CurrentTable}
WHERE cve_id = ANY(@CveIds)
ORDER BY cve_id
""";
await using var connection = await _dataSource.OpenSystemConnectionAsync(cancellationToken).ConfigureAwait(false);
var rows = await connection.QueryAsync<CurrentRow>(new CommandDefinition(
sql,
new { CveIds = normalized },
cancellationToken: cancellationToken)).ConfigureAwait(false);
var result = new Dictionary<string, EpssCurrentEntry>(StringComparer.Ordinal);
foreach (var row in rows)
{
result[row.cve_id] = new EpssCurrentEntry(
row.cve_id,
(double)row.epss_score,
(double)row.percentile,
row.model_date,
row.import_run_id);
}
return result;
}
public async Task<IReadOnlyList<EpssHistoryEntry>> GetHistoryAsync(
string cveId,
int days,
CancellationToken cancellationToken = default)
{
ArgumentException.ThrowIfNullOrWhiteSpace(cveId);
var normalized = cveId.Trim().ToUpperInvariant();
var limit = Math.Clamp(days, 1, 3650);
var sql = $"""
SELECT model_date, epss_score, percentile, import_run_id
FROM {ScoresTable}
WHERE cve_id = @CveId
ORDER BY model_date DESC
LIMIT @Limit
""";
await using var connection = await _dataSource.OpenSystemConnectionAsync(cancellationToken).ConfigureAwait(false);
var rows = await connection.QueryAsync<HistoryRow>(new CommandDefinition(
sql,
new { CveId = normalized, Limit = limit },
cancellationToken: cancellationToken)).ConfigureAwait(false);
return rows.Select(static row => new EpssHistoryEntry(
row.model_date,
(double)row.epss_score,
(double)row.percentile,
row.import_run_id))
.ToList();
}
private static async Task EnsurePartitionsAsync(
NpgsqlConnection connection,
NpgsqlTransaction transaction,
DateOnly modelDate,
CancellationToken cancellationToken)
{
var sql = "SELECT create_epss_partition(@Year, @Month)";
await connection.ExecuteAsync(new CommandDefinition(
sql,
new { Year = modelDate.Year, Month = modelDate.Month },
transaction: transaction,
cancellationToken: cancellationToken)).ConfigureAwait(false);
}
private static async Task<(int RowCount, int DistinctCount)> CopyStageAsync(
NpgsqlConnection connection,
NpgsqlTransaction transaction,
string stageTable,
IAsyncEnumerable<EpssScoreRow> rows,
CancellationToken cancellationToken)
{
var rowCount = 0;
await using (var importer = connection.BeginBinaryImport($"COPY {stageTable} (cve_id, epss_score, percentile) FROM STDIN (FORMAT BINARY)"))
{
await foreach (var row in rows.WithCancellation(cancellationToken).ConfigureAwait(false))
{
await importer.StartRowAsync(cancellationToken).ConfigureAwait(false);
await importer.WriteAsync(row.CveId, NpgsqlDbType.Text, cancellationToken).ConfigureAwait(false);
await importer.WriteAsync(row.Score, NpgsqlDbType.Double, cancellationToken).ConfigureAwait(false);
await importer.WriteAsync(row.Percentile, NpgsqlDbType.Double, cancellationToken).ConfigureAwait(false);
rowCount++;
}
await importer.CompleteAsync(cancellationToken).ConfigureAwait(false);
}
var countsSql = $"""
SELECT COUNT(*) AS total, COUNT(DISTINCT cve_id) AS distinct_count
FROM {stageTable}
""";
var counts = await connection.QuerySingleAsync<StageCounts>(new CommandDefinition(
countsSql,
transaction: transaction,
cancellationToken: cancellationToken)).ConfigureAwait(false);
return (rowCount, counts.distinct_count);
}
private async Task InsertChangesAsync(
NpgsqlConnection connection,
NpgsqlTransaction transaction,
string stageTable,
DateOnly modelDate,
Guid importRunId,
CancellationToken cancellationToken)
{
var sql = $"""
INSERT INTO {ChangesTable} (
model_date,
cve_id,
old_score,
new_score,
delta_score,
old_percentile,
new_percentile,
delta_percentile,
flags,
import_run_id
)
SELECT
@ModelDate,
s.cve_id,
c.epss_score AS old_score,
s.epss_score AS new_score,
CASE WHEN c.epss_score IS NULL THEN NULL ELSE s.epss_score - c.epss_score END AS delta_score,
c.percentile AS old_percentile,
s.percentile AS new_percentile,
CASE WHEN c.percentile IS NULL THEN NULL ELSE s.percentile - c.percentile END AS delta_percentile,
compute_epss_change_flags(
c.epss_score,
s.epss_score,
c.percentile,
s.percentile,
cfg.high_score,
cfg.high_percentile,
cfg.big_jump_delta
) AS flags,
@ImportRunId
FROM {stageTable} s
LEFT JOIN {CurrentTable} c ON c.cve_id = s.cve_id
CROSS JOIN (
SELECT high_score, high_percentile, big_jump_delta
FROM {ConfigTable}
WHERE org_id IS NULL
LIMIT 1
) cfg
""";
await connection.ExecuteAsync(new CommandDefinition(
sql,
new { ModelDate = modelDate, ImportRunId = importRunId },
transaction: transaction,
cancellationToken: cancellationToken)).ConfigureAwait(false);
}
private async Task UpsertCurrentAsync(
NpgsqlConnection connection,
NpgsqlTransaction transaction,
string stageTable,
DateOnly modelDate,
Guid importRunId,
DateTimeOffset updatedAtUtc,
CancellationToken cancellationToken)
{
var sql = $"""
INSERT INTO {CurrentTable} (
cve_id,
epss_score,
percentile,
model_date,
import_run_id,
updated_at
)
SELECT
cve_id,
epss_score,
percentile,
@ModelDate,
@ImportRunId,
@UpdatedAtUtc
FROM {stageTable}
ON CONFLICT (cve_id) DO UPDATE SET
epss_score = EXCLUDED.epss_score,
percentile = EXCLUDED.percentile,
model_date = EXCLUDED.model_date,
import_run_id = EXCLUDED.import_run_id,
updated_at = EXCLUDED.updated_at
""";
await connection.ExecuteAsync(new CommandDefinition(
sql,
new { ModelDate = modelDate, ImportRunId = importRunId, UpdatedAtUtc = updatedAtUtc },
transaction: transaction,
cancellationToken: cancellationToken)).ConfigureAwait(false);
}
private sealed class StageCounts
{
public int distinct_count { get; set; }
}
private sealed class ImportRunRow
{
public Guid import_run_id { get; set; }
public DateOnly model_date { get; set; }
public string source_uri { get; set; } = "";
public DateTimeOffset retrieved_at { get; set; }
public string file_sha256 { get; set; } = "";
public string? decompressed_sha256 { get; set; }
public int row_count { get; set; }
public string? model_version_tag { get; set; }
public DateOnly? published_date { get; set; }
public string status { get; set; } = "";
public string? error { get; set; }
public DateTimeOffset created_at { get; set; }
public EpssImportRun ToModel() => new(
ImportRunId: import_run_id,
ModelDate: model_date,
SourceUri: source_uri,
RetrievedAtUtc: retrieved_at,
FileSha256: file_sha256,
DecompressedSha256: decompressed_sha256,
RowCount: row_count,
ModelVersionTag: model_version_tag,
PublishedDate: published_date,
Status: status,
Error: error,
CreatedAtUtc: created_at);
}
private sealed class CurrentRow
{
public string cve_id { get; set; } = "";
public decimal epss_score { get; set; }
public decimal percentile { get; set; }
public DateOnly model_date { get; set; }
public Guid import_run_id { get; set; }
}
private sealed class HistoryRow
{
public DateOnly model_date { get; set; }
public decimal epss_score { get; set; }
public decimal percentile { get; set; }
public Guid import_run_id { get; set; }
}
private static void EnsureTypeHandlers()
{
if (Interlocked.Exchange(ref _typeHandlersRegistered, 1) == 1)
{
return;
}
SqlMapper.AddTypeHandler(new DateOnlyTypeHandler());
SqlMapper.AddTypeHandler(new NullableDateOnlyTypeHandler());
}
private sealed class DateOnlyTypeHandler : SqlMapper.TypeHandler<DateOnly>
{
public override void SetValue(IDbDataParameter parameter, DateOnly value)
{
parameter.Value = value;
if (parameter is NpgsqlParameter npgsqlParameter)
{
npgsqlParameter.NpgsqlDbType = NpgsqlDbType.Date;
}
}
public override DateOnly Parse(object value)
{
return value switch
{
DateOnly dateOnly => dateOnly,
DateTime dateTime => DateOnly.FromDateTime(dateTime),
_ => DateOnly.FromDateTime((DateTime)value)
};
}
}
private sealed class NullableDateOnlyTypeHandler : SqlMapper.TypeHandler<DateOnly?>
{
public override void SetValue(IDbDataParameter parameter, DateOnly? value)
{
if (value is null)
{
parameter.Value = DBNull.Value;
return;
}
parameter.Value = value.Value;
if (parameter is NpgsqlParameter npgsqlParameter)
{
npgsqlParameter.NpgsqlDbType = NpgsqlDbType.Date;
}
}
public override DateOnly? Parse(object value)
{
if (value is null || value is DBNull)
{
return null;
}
return value switch
{
DateOnly dateOnly => dateOnly,
DateTime dateTime => DateOnly.FromDateTime(dateTime),
_ => DateOnly.FromDateTime((DateTime)value)
};
}
}
}

89
src/Scanner/__Libraries/StellaOps.Scanner.Storage/Repositories/IEpssRepository.cs Normal file
View File

@@ -0,0 +1,89 @@
// -----------------------------------------------------------------------------
// IEpssRepository.cs
// Sprint: SPRINT_3410_0001_0001_epss_ingestion_storage
// Tasks: EPSS-3410-007, EPSS-3410-008
// Description: EPSS persistence contract (import runs, scores/current projection, change log).
// -----------------------------------------------------------------------------
using StellaOps.Scanner.Storage.Epss;
namespace StellaOps.Scanner.Storage.Repositories;
public interface IEpssRepository
{
Task<EpssImportRun?> GetImportRunAsync(DateOnly modelDate, CancellationToken cancellationToken = default);
/// <summary>
/// Creates (or resets) the import run record for a model date.
/// </summary>
Task<EpssImportRun> BeginImportAsync(
DateOnly modelDate,
string sourceUri,
DateTimeOffset retrievedAtUtc,
string fileSha256,
CancellationToken cancellationToken = default);
Task MarkImportSucceededAsync(
Guid importRunId,
int rowCount,
string? decompressedSha256,
string? modelVersionTag,
DateOnly? publishedDate,
CancellationToken cancellationToken = default);
Task MarkImportFailedAsync(
Guid importRunId,
string error,
CancellationToken cancellationToken = default);
/// <summary>
/// Writes the EPSS snapshot into time-series storage, computes changes, and updates the current projection.
/// </summary>
Task<EpssWriteResult> WriteSnapshotAsync(
Guid importRunId,
DateOnly modelDate,
DateTimeOffset updatedAtUtc,
IAsyncEnumerable<EpssScoreRow> rows,
CancellationToken cancellationToken = default);
Task<IReadOnlyDictionary<string, EpssCurrentEntry>> GetCurrentAsync(
IEnumerable<string> cveIds,
CancellationToken cancellationToken = default);
Task<IReadOnlyList<EpssHistoryEntry>> GetHistoryAsync(
string cveId,
int days,
CancellationToken cancellationToken = default);
}
public sealed record EpssImportRun(
Guid ImportRunId,
DateOnly ModelDate,
string SourceUri,
DateTimeOffset RetrievedAtUtc,
string FileSha256,
string? DecompressedSha256,
int RowCount,
string? ModelVersionTag,
DateOnly? PublishedDate,
string Status,
string? Error,
DateTimeOffset CreatedAtUtc);
public readonly record struct EpssWriteResult(
int RowCount,
int DistinctCveCount);
public sealed record EpssCurrentEntry(
string CveId,
double Score,
double Percentile,
DateOnly ModelDate,
Guid ImportRunId);
public sealed record EpssHistoryEntry(
DateOnly ModelDate,
double Score,
double Percentile,
Guid ImportRunId);

11
src/Scanner/__Libraries/StellaOps.Scanner.Storage/TASKS.md
View File

@@ -3,4 +3,13 @@
| Task ID | Sprint | Status | Notes |
| --- | --- | --- | --- |
| `PROOFSPINE-3100-DB` | `docs/implplan/SPRINT_3100_0001_0001_proof_spine_system.md` | DOING | Add Postgres migrations and repository for ProofSpine persistence (`proof_spines`, `proof_segments`, `proof_spine_history`). |
| `SCAN-API-3103-004` | `docs/implplan/SPRINT_3103_0001_0001_scanner_api_ingestion_completion.md` | DOING | Fix scanner storage connection/schema issues surfaced by Scanner WebService ingestion tests. |
| `SCAN-API-3103-004` | `docs/implplan/SPRINT_3103_0001_0001_scanner_api_ingestion_completion.md` | DONE | Fix scanner storage connection/schema issues surfaced by Scanner WebService ingestion tests. |
| `DRIFT-3600-DB` | `docs/implplan/SPRINT_3600_0003_0001_drift_detection_engine.md` | DONE | Add drift tables migration + code change/drift result repositories + DI wiring. |
| `EPSS-3410-001` | `docs/implplan/SPRINT_3410_0001_0001_epss_ingestion_storage.md` | DONE | Added EPSS schema migration `Postgres/Migrations/008_epss_integration.sql` and wired via `MigrationIds.cs`. |
| `EPSS-3410-002` | `docs/implplan/SPRINT_3410_0001_0001_epss_ingestion_storage.md` | DOING | Implement `EpssScoreRow` + ingestion models. |
| `EPSS-3410-003` | `docs/implplan/SPRINT_3410_0001_0001_epss_ingestion_storage.md` | DOING | Implement `IEpssSource` interface (online vs bundle). |
| `EPSS-3410-004` | `docs/implplan/SPRINT_3410_0001_0001_epss_ingestion_storage.md` | DOING | Implement `EpssOnlineSource` (download to temp; hash provenance). |
| `EPSS-3410-005` | `docs/implplan/SPRINT_3410_0001_0001_epss_ingestion_storage.md` | DOING | Implement `EpssBundleSource` (air-gap file input). |
| `EPSS-3410-006` | `docs/implplan/SPRINT_3410_0001_0001_epss_ingestion_storage.md` | DOING | Implement streaming `EpssCsvStreamParser` (validation + header comment extraction). |
| `EPSS-3410-007` | `docs/implplan/SPRINT_3410_0001_0001_epss_ingestion_storage.md` | DOING | Implement Postgres `IEpssRepository` (runs + scores/current/changes). |
| `EPSS-3410-008` | `docs/implplan/SPRINT_3410_0001_0001_epss_ingestion_storage.md` | DOING | Implement change detection + flags (`compute_epss_change_flags` + delta join). |