From 10b07e9aae1ddd993878327d13763f06922554ad Mon Sep 17 00:00:00 2001 From: CHILING12 <940580717@qq.com> Date: Sat, 5 Sep 2026 09:26:49 +0800 Subject: [PATCH] Add RITA-Lab reporting layer --- CHANGELOG_RITA_LAB.md | 258 +++++++++++++++++++++++ Makefile | 14 +- README.md | 3 + cmd/cmd.go | 1 + cmd/lab.go | 190 +++++++++++++++++ cmd/lab_test.go | 27 +++ docker-compose.lab.yml | 68 ++++++ docs/Configuration.md | 6 + docs/RITALab.md | 127 +++++++++++ lab/aggregate.go | 143 +++++++++++++ lab/allowlist.go | 29 +++ lab/assets.go | 31 +++ lab/config.go | 98 +++++++++ lab/dns.go | 81 ++++++++ lab/evaluate.go | 22 ++ lab/fixtures/config/lab-profile.hjson | 47 +++++ lab/fixtures/config/rita-lab.hjson | 12 ++ lab/fixtures/manifest.hjson | 29 +++ lab/lab_test.go | 102 +++++++++ lab/model.go | 186 +++++++++++++++++ lab/normalize.go | 65 ++++++ lab/query.go | 289 ++++++++++++++++++++++++++ lab/report.go | 136 ++++++++++++ lab/report_test.go | 44 ++++ lab/run.go | 129 ++++++++++++ lab/scoring.go | 133 ++++++++++++ 26 files changed, 2269 insertions(+), 1 deletion(-) create mode 100644 CHANGELOG_RITA_LAB.md create mode 100644 cmd/lab.go create mode 100644 cmd/lab_test.go create mode 100644 docker-compose.lab.yml create mode 100644 docs/RITALab.md create mode 100644 lab/aggregate.go create mode 100644 lab/allowlist.go create mode 100644 lab/assets.go create mode 100644 lab/config.go create mode 100644 lab/dns.go create mode 100644 lab/evaluate.go create mode 100644 lab/fixtures/config/lab-profile.hjson create mode 100644 lab/fixtures/config/rita-lab.hjson create mode 100644 lab/fixtures/manifest.hjson create mode 100644 lab/lab_test.go create mode 100644 lab/model.go create mode 100644 lab/normalize.go create mode 100644 lab/query.go create mode 100644 lab/report.go create mode 100644 lab/report_test.go create mode 100644 lab/run.go create mode 100644 lab/scoring.go diff --git a/CHANGELOG_RITA_LAB.md b/CHANGELOG_RITA_LAB.md new file mode 100644 index 0000000..6d14b41 --- /dev/null +++ b/CHANGELOG_RITA_LAB.md @@ -0,0 +1,258 @@ +# RITA-Lab 改动总结 + +> 本文件总结本次在 RITA v5 checkout 中完成的全部 RITA-Lab 相关改动。RITA-Lab 是只读的报告与实验层，不改变 RITA 原生检测算法、原生评分语义或 `threat_mixtape` 表结构。 + +## 1. 目标与边界 + +- 在现有 RITA CLI 中内置 `rita lab`，而不是创建平行 CLI。 +- 以仓库内已有 Zeek 日志 fixture 作为首版离线、可复现输入。 +- 仅消费 RITA 导入和分析后的 ClickHouse 结果，生成资产感知、可解释的二次报告。 +- 不抓取宿主机接口、不扫描公网、不联系外部目标、不生成攻击流量。 +- 报告期 allowlist 只影响二次报告优先级，不映射到 RITA 的导入期 `never_included_domains`，从而保留可审计证据。 + +## 2. 核心领域模型 + +新增 `lab/model.go`，定义并区分： + +- `LabConfig`、`Asset`、`Allowlists`、`AllowlistRule`； +- `ScoringConfig`、`ReportingConfig`、`DNSFeatureConfig`； +- `NativeEvidence`、`DNSEvent`、`DNSFeatures`； +- `AssetMatch`、`AllowlistMatch`； +- `AggregateAlert`、`ScoreContribution`、`Report`、`EvaluationResult`。 + +支持的检测类型： + +- `beacon` +- `long_connection` +- `strobe` +- `dns_c2` +- `threat_intel_only` + +缺少域名时使用 `ip:` 作为明确的 IP fallback；缺少来源时保留 `unattributed`，不伪造主机地址。 + +## 3. 独立实验室配置 + +新增 `lab/config.go` 和 fixture 配置： + +- `lab/fixtures/config/rita-lab.hjson`：离线 RITA 导入配置、内部网段和关闭更新检查； +- `lab/fixtures/config/lab-profile.hjson`：资产、allowlist、评分权重、默认窗口和 DNS 特征开关。 + +配置读取和校验包括： + +- HJSON 解析和配置 SHA-256； +- schema 版本和默认时间窗口； +- 资产 ID、CIDR、重复 CIDR、资产重要性； +- allowlist 模式、原因和分数扣减； +- 五项评分权重之和为 1； +- persistence 阈值有效性。 + +## 4. 资产与域名关联 + +新增： + +- `lab/assets.go`：使用最长前缀匹配将源 IP 关联到最具体资产；未匹配资产保留为 `unclassified`； +- `lab/allowlist.go`：支持精确域名和 `*.` 通配符，统一小写并移除尾点； +- `lab/normalize.go`：统一域名、时间窗口、目标键和稳定告警 ID。 + +## 5. ClickHouse 原生证据查询 + +新增 `lab/query.go`： + +- 从 `threat_mixtape` 读取报告范围内每个 hash 的最新基线快照； +- 快照严格关联 `hash`、`import_id`、`last_seen` 和 `analyzed_at`，避免历史结果或 modifier 行重复计入； +- modifier 单独读取并汇总到对应基线证据； +- native final score 对齐 RITA viewer 的原生组成，包括 prevalence、first-seen、missing-host-header、威胁情报数据量和 DNS direct-connection 分量； +- 查询再次施加 `[from,to)` 时间范围； +- DNS 候选域使用 ClickHouse `Array(String)` 参数，并安全转义参数值； +- DNS 域名查询统一处理大小写和尾点。 + +## 6. DNS C2 证据和安全归因 + +新增 `lab/dns.go` 和相关聚合逻辑： + +- 查询数； +- 唯一编码标签数； +- 平均和最大编码标签长度； +- 以字节为单位计算 Shannon entropy（bits/byte）； +- 每分钟查询频率； +- NXDOMAIN 数量和比例； +- 首次和最后观测时间。 + +重要语义：RITA 的 DNS C2 结果可能是域级证据且来源为 `::`。RITA-Lab 将其保留为明确的未归因域级 alert；raw DNS 的 `dns.src` 只作为调查上下文，绝不把域级分数复制给该域的每个查询主机，避免错误提升普通查询主机的风险分数。 + +当没有 DNS 事件、编码标签或有效分母时，相关特征保持不可用，而不是被解释为低风险。 + +## 7. 时间窗口聚合 + +新增 `lab/aggregate.go`： + +- 使用以下逻辑键聚合： + + ```text + source_ip + destination_domain + detection_type + time_window + ``` + +- 同一个 RITA 原生结果可以展开为多个明确检测类型； +- 保存原生证据、连接计数、DNS 查询计数、首末时间、威胁情报状态、资产和 allowlist 信息； +- 按实验室优先级、窗口、源、目标、检测类型和 ID 进行确定性排序； +- DNS 特征开关关闭时不执行 raw DNS 查询。 + +## 8. 二次优先级与解释 + +新增 `lab/scoring.go`。默认使用归一化到 `0..1` 的组件： + +```text +pre_allowlist_score = + 0.45 * rita_evidence + + 0.20 * asset_importance + + 0.15 * threat_intel + + 0.10 * persistence + + 0.10 * rarity + +lab_priority_score = 100 * clamp(pre_allowlist_score - allowlist_reduction, 0, 1) +``` + +每项都输出： + +- 原始值； +- 归一化值； +- 配置权重； +- 有符号贡献； +- `applied` 或 `not_available` 状态； +- 可读原因。 + +未知 prevalence 和未分类资产不会被虚构成正向风险贡献。allowlist 的匹配规则、原因和扣减作为独立的可审计贡献保留。 + +## 9. 报告输出 + +新增 `lab/report.go`： + +- Markdown：元数据、摘要、原生证据、资产、DNS 特征、allowlist 和评分分解； +- CSV：使用 Go `encoding/csv`，包含结构化字段、allowlist 原因和 JSON `score_breakdown_json`； +- HTML：使用 `html/template`，对域名、owner、allowlist reason 和日志值进行 HTML 转义； +- 所有格式共用同一 `Report` 模型和确定性排序； +- 记录配置 SHA-256 和评分版本； +- 评估指标仅记录实际执行得到的导入计数、聚合告警数和耗时，不虚构准确率、召回率或性能数据。 + +`lab/run.go` 还提供： + +- 报告构建； +- 最低优先级筛选； +- allowlist 告警默认保留、可选择排除； +- Markdown/CSV/HTML 格式解析； +- 输出文件预检查； +- 临时文件写入后原子 rename，避免多格式输出中途失败留下半成品。 + +## 10. CLI + +新增 `cmd/lab.go`，并在 `cmd/cmd.go` 注册 `LabCommand`。 + +### `rita lab report` + +支持： + +- `--config` +- `--lab-config` +- `--from` / `--to` +- `--window` +- `--formats` +- `--output-dir` +- `--minimum-score` +- `--exclude-allowlisted` +- `--overwrite` + +### `rita lab evaluate` + +支持： + +- `--logs` 受控日志目录； +- `--rebuild` 显式重建数据库； +- 与 report 相同的范围、窗口、格式、输出和 allowlist 选项； +- 使用既有 `RunImportCmd`，不复制 RITA importer、analysis 或 modifier 流程； +- 执行前校验日志目录。 + +省略 `--from` 和 `--to` 时复用 RITA viewer 的近期范围辅助函数，该函数会将下界限制在最新数据时间戳前 24 小时，而非宣称覆盖整个历史数据集。 + +## 11. Fixture、Compose 和 Makefile + +新增： + +- `lab/fixtures/manifest.hjson`：登记已有 `dnscat2-ja3-strobe-agent`、`valid_tsv` 和 `dns_only` fixture；未执行的指标明确标记为 `not_measured`； +- `docker-compose.lab.yml`：ClickHouse、RITA-Lab runner 和可选离线 Zeek PCAP profile；使用 internal-only network、无端口发布和只读 fixture 挂载； +- `Makefile` 目标： + - `test-lab` + - `compose-lab-config` + - `lab-up` + - `lab-down` + +可选 PCAP profile 仅读取用户提供的本地 PCAP，不绑定宿主网络接口，也不对公网通信。 + +## 12. 测试与验证 + +新增测试覆盖： + +- 资产最长前缀和未分类资产； +- allowlist 大小写、尾点和通配符； +- DNS 熵、标签长度、NXDOMAIN 和频率； +- 未归因 DNS C2 保留域级证据； +- 时间窗口和稳定告警 ID； +- allowlist 降分但保留告警； +- CSV 特殊字符转义； +- HTML 内容转义； +- 报告格式解析； +- CLI 注册和窗口校验。 + +在本环境中完成： + +```text +go build ./... 通过 +go test ./lab 通过 +go test ./cmd -run '^TestLab' 通过 +git diff --check 通过 +``` + +完整测试套件未能在当前环境完成，原因是： + +- 没有 Docker daemon，依赖 Testcontainers 的集成测试无法启动 ClickHouse； +- Docker 命令本身也不可用，因此 Compose 未实际启动； +- 原项目的 GitHub 更新检查测试受到 GitHub API rate limit 影响。 + +Go 1.22.12 使用的是本地临时工具链 `/tmp/go`，不是系统级安装。 + +## 13. 变更文件清单 + +### 修改的既有文件 + +- `Makefile` +- `README.md` +- `cmd/cmd.go` +- `docs/Configuration.md` + +### 新增文件 + +- `CHANGELOG_RITA_LAB.md` +- `cmd/lab.go` +- `cmd/lab_test.go` +- `docker-compose.lab.yml` +- `docs/RITALab.md` +- `lab/model.go` +- `lab/config.go` +- `lab/normalize.go` +- `lab/assets.go` +- `lab/allowlist.go` +- `lab/dns.go` +- `lab/aggregate.go` +- `lab/scoring.go` +- `lab/query.go` +- `lab/report.go` +- `lab/run.go` +- `lab/evaluate.go` +- `lab/lab_test.go` +- `lab/report_test.go` +- `lab/fixtures/config/rita-lab.hjson` +- `lab/fixtures/config/lab-profile.hjson` +- `lab/fixtures/manifest.hjson` + +## 14. 安全声明 + +本次实现遵守实验室边界：**不要对公网进行扫描或攻击测试。** 所有默认实验路径均为仓库内受控日志回放或用户明确提供的本地离线 PCAP 处理。 diff --git a/Makefile b/Makefile index bb916f4..7ea2ce1 100644 --- a/Makefile +++ b/Makefile @@ -5,7 +5,7 @@ CGO_ENABLED ?= 0 GOARCH ?= $(shell go env GOARCH) GOOS ?= $(shell go env GOOS) -.PHONY: build test test-unit test-integration test-database test-cmd test-viewer clean +.PHONY: build test test-unit test-integration test-database test-cmd test-viewer test-lab compose-lab-config lab-up lab-down clean build: CGO_ENABLED=$(CGO_ENABLED) GOARCH=$(GOARCH) GOOS=$(GOOS) go build $(LDFLAGS) -o rita @@ -27,5 +27,17 @@ test-cmd: test-viewer: go test ./viewer/... -timeout 1800s +test-lab: + go test ./lab/... + +compose-lab-config: + docker compose -f docker-compose.lab.yml config + +lab-up: + docker compose -f docker-compose.lab.yml up --build --abort-on-container-exit + +lab-down: + docker compose -f docker-compose.lab.yml down --volumes + clean: rm -f rita diff --git a/README.md b/README.md index 9bafb2a..9ca6896 100644 --- a/README.md +++ b/README.md @@ -66,6 +66,9 @@ To destroy and recreate a dataset, use the `--rebuild` flag. ## Configuration See [Configuration](/docs/Configuration.md) for details on adjusting scoring. +## RITA-Lab reports +This checkout also includes an optional, read-only RITA-Lab reporting layer for controlled Zeek-log experiments. It adds asset association, report-period allowlist annotations, time-window alert aggregation, explainable secondary triage priority, and Markdown/CSV/HTML output without changing RITA's native detection logic. See [RITA-Lab](docs/RITALab.md) for the offline fixture workflow and safety boundary. + ## Searching RITA follows a GitHub-style search syntax. Each field follows the `:` format, with each search criteria separated by a space. diff --git a/cmd/cmd.go b/cmd/cmd.go index 03e964a..c9d200a 100644 --- a/cmd/cmd.go +++ b/cmd/cmd.go @@ -23,6 +23,7 @@ func Commands() []*cli.Command { return []*cli.Command{ ImportCommand, ViewCommand, + LabCommand, DeleteCommand, ListCommand, ValidateConfigCommand, diff --git a/cmd/lab.go b/cmd/lab.go new file mode 100644 index 0000000..0ad5d3f --- /dev/null +++ b/cmd/lab.go @@ -0,0 +1,190 @@ +package cmd + +import ( + "context" + "errors" + "fmt" + "os" + "strings" + "time" + + "github.com/activecm/rita/v5/config" + "github.com/activecm/rita/v5/database" + "github.com/activecm/rita/v5/lab" + "github.com/spf13/afero" + "github.com/urfave/cli/v2" +) + +var ( + ErrLabDatabaseRequired = errors.New("database name is required") + ErrLabTimeRange = errors.New("--from and --to must be specified together, or both omitted") + ErrLabMinimumScore = errors.New("minimum score must be between 0 and 100") +) + +var LabCommand = &cli.Command{ + Name: "lab", + Usage: "generate asset-aware laboratory reports from RITA analysis data", + Subcommands: []*cli.Command{ + labReportCommand, + labEvaluateCommand, + }, +} + +var labReportCommand = &cli.Command{ + Name: "report", + Usage: "write explainable Markdown, CSV, and HTML laboratory reports", + UsageText: "rita lab report [options]", + Flags: append([]cli.Flag{ + &cli.StringFlag{Name: "lab-config", Usage: "load laboratory configuration from FILE", Required: true}, + &cli.StringFlag{Name: "from", Usage: "inclusive UTC RFC3339 report start"}, + &cli.StringFlag{Name: "to", Usage: "exclusive UTC RFC3339 report end"}, + &cli.StringFlag{Name: "window", Usage: "aggregation window duration", Value: ""}, + &cli.StringFlag{Name: "formats", Usage: "comma-separated: markdown,csv,html", Value: "markdown,csv,html"}, + &cli.StringFlag{Name: "output-dir", Usage: "directory for generated reports", Value: "./rita-lab-output"}, + &cli.Float64Flag{Name: "minimum-score", Usage: "minimum laboratory priority (0-100)", Value: 0}, + &cli.BoolFlag{Name: "exclude-allowlisted", Usage: "omit alerts that match the laboratory allowlist", Value: false}, + &cli.BoolFlag{Name: "overwrite", Usage: "replace an existing report file", Value: false}, + }, ConfigFlag(false)), + Action: runLabReport, +} + +var labEvaluateCommand = &cli.Command{ + Name: "evaluate", + Usage: "run a controlled Zeek-log import and emit actual report measurements", + UsageText: "rita lab evaluate --logs DIRECTORY [options]", + Flags: append([]cli.Flag{ + &cli.StringFlag{Name: "lab-config", Usage: "load laboratory configuration from FILE", Required: true}, + &cli.StringFlag{Name: "logs", Usage: "controlled Zeek log directory", Required: true}, + &cli.StringFlag{Name: "from", Usage: "inclusive UTC RFC3339 report start"}, + &cli.StringFlag{Name: "to", Usage: "exclusive UTC RFC3339 report end"}, + &cli.StringFlag{Name: "window", Usage: "aggregation window duration", Value: ""}, + &cli.StringFlag{Name: "formats", Usage: "comma-separated: markdown,csv,html", Value: "markdown,csv,html"}, + &cli.StringFlag{Name: "output-dir", Usage: "directory for generated reports", Value: "./rita-lab-output"}, + &cli.BoolFlag{Name: "exclude-allowlisted", Usage: "omit alerts that match the laboratory allowlist", Value: false}, + &cli.BoolFlag{Name: "rebuild", Usage: "destroy and rebuild the named experiment database", Value: false}, + &cli.BoolFlag{Name: "overwrite", Usage: "replace an existing report file", Value: false}, + }, ConfigFlag(false)), + Action: runLabEvaluate, +} + +func runLabReport(cCtx *cli.Context) error { + return runLabReportWithTiming(cCtx, nil, time.Time{}) +} + +func runLabEvaluate(cCtx *cli.Context) error { + if !cCtx.Args().Present() { + return ErrLabDatabaseRequired + } + if err := ValidateDatabaseName(cCtx.Args().First()); err != nil { + return err + } + fs := afero.NewOsFs() + if err := ValidateLogDirectory(fs, cCtx.String("logs")); err != nil { + return err + } + cfg, err := config.ReadFileConfig(fs, cCtx.String("config")) + if err != nil { + return err + } + started := time.Now() + results, err := RunImportCmd(started, cfg, fs, cCtx.String("logs"), cCtx.Args().First(), false, cCtx.Bool("rebuild")) + if err != nil { + return err + } + return runLabReportWithTiming(cCtx, &results, started) +} + +func runLabReportWithTiming(cCtx *cli.Context, importResults *ImportResults, importStartedAt time.Time) error { + if !cCtx.Args().Present() { + return ErrLabDatabaseRequired + } + databaseName := cCtx.Args().First() + if err := ValidateDatabaseName(databaseName); err != nil { + return err + } + if cCtx.Float64("minimum-score") < 0 || cCtx.Float64("minimum-score") > 100 { + return ErrLabMinimumScore + } + + fs := afero.NewOsFs() + cfg, err := config.ReadFileConfig(fs, cCtx.String("config")) + if err != nil { + return err + } + labCfg, configHash, err := lab.ReadConfig(cCtx.String("lab-config")) + if err != nil { + return err + } + window, err := labWindow(cCtx.String("window"), labCfg.Reporting.DefaultWindow) + if err != nil { + return err + } + formats, err := lab.ParseFormats(cCtx.String("formats")) + if err != nil { + return err + } + + db, err := database.ConnectToDB(context.Background(), databaseName, cfg, nil) + if err != nil { + return err + } + defer db.Conn.Close() + from, to, err := labTimeRange(cCtx.String("from"), cCtx.String("to"), db) + if err != nil { + return err + } + + started := time.Now() + report, err := lab.BuildReport(db, labCfg, configHash, from, to, window) + if err != nil { + return err + } + report = lab.FilterReport(report, cCtx.Float64("minimum-score"), !cCtx.Bool("exclude-allowlisted")) + if importResults != nil { + report = lab.WithEvaluation(report, lab.NewEvaluationResult(importResults.ResultCounts, importStartedAt, time.Since(started), len(report.Alerts))) + } + paths, err := lab.WriteReportFiles(report, cCtx.String("output-dir"), formats, cCtx.Bool("overwrite")) + if err != nil { + return err + } + for _, path := range paths { + fmt.Fprintln(os.Stdout, path) + } + return nil +} + +func labWindow(value, fallback string) (time.Duration, error) { + if value == "" { + value = fallback + } + window, err := time.ParseDuration(value) + if err != nil || window <= 0 { + return 0, fmt.Errorf("invalid --window %q", value) + } + return window, nil +} + +func labTimeRange(fromString, toString string, db *database.DB) (time.Time, time.Time, error) { + if (fromString == "") != (toString == "") { + return time.Time{}, time.Time{}, ErrLabTimeRange + } + if fromString == "" { + from, to, _, _, err := db.GetTrueMinMaxTimestamps() + if err != nil { + return time.Time{}, time.Time{}, err + } + return from.UTC(), to.UTC().Add(time.Second), nil + } + from, err := time.Parse(time.RFC3339, strings.TrimSpace(fromString)) + if err != nil { + return time.Time{}, time.Time{}, fmt.Errorf("parse --from: %w", err) + } + to, err := time.Parse(time.RFC3339, strings.TrimSpace(toString)) + if err != nil { + return time.Time{}, time.Time{}, fmt.Errorf("parse --to: %w", err) + } + if !from.Before(to) { + return time.Time{}, time.Time{}, ErrLabTimeRange + } + return from.UTC(), to.UTC(), nil +} diff --git a/cmd/lab_test.go b/cmd/lab_test.go new file mode 100644 index 0000000..2430ddc --- /dev/null +++ b/cmd/lab_test.go @@ -0,0 +1,27 @@ +package cmd + +import ( + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +func TestLabCommandIsRegistered(t *testing.T) { + var found bool + for _, command := range Commands() { + if command.Name == "lab" { + found = true + break + } + } + require.True(t, found) +} + +func TestLabWindow(t *testing.T) { + window, err := labWindow("", "30m") + require.NoError(t, err) + require.Equal(t, 30*time.Minute, window) + _, err = labWindow("0m", "30m") + require.Error(t, err) +} diff --git a/docker-compose.lab.yml b/docker-compose.lab.yml new file mode 100644 index 0000000..be1d8ef --- /dev/null +++ b/docker-compose.lab.yml @@ -0,0 +1,68 @@ +# Offline, reproducible RITA-Lab replay. It imports repository-owned Zeek logs; +# it does not capture a host interface, scan, or send traffic to the Internet. +name: rita-lab + +services: + clickhouse: + image: clickhouse/clickhouse-server:${CLICKHOUSE_VERSION:-24.1.6} + environment: + CLICKHOUSE_DB: default + CLICKHOUSE_USER: default + CLICKHOUSE_PASSWORD: "" + healthcheck: + test: ["CMD-SHELL", "wget --no-verbose --tries=1 --spider http://localhost:8123/ping || exit 1"] + interval: 3s + timeout: 3s + retries: 30 + volumes: + - lab-clickhouse:/var/lib/clickhouse + - ./deployment/config.xml:/etc/clickhouse-server/users.d/custom_config.xml:ro + - ./deployment/timezone.xml:/etc/clickhouse-server/config.d/timezone.xml:ro + networks: [lab] + + rita-lab: + build: . + depends_on: + clickhouse: + condition: service_healthy + environment: + DB_ADDRESS: clickhouse:9000 + CLICKHOUSE_USERNAME: default + CLICKHOUSE_PASSWORD: "" + CONFIG_DIR: /deployment + LOG_LEVEL: "1" + APP_ENV: production + volumes: + - ./.env:/.env:ro + - ./deployment:/deployment:ro + - ./lab/fixtures/config/rita-lab.hjson:/rita-lab/rita-lab.hjson:ro + - ./lab/fixtures/config/lab-profile.hjson:/rita-lab/lab-profile.hjson:ro + - ./test_data/dnscat2-ja3-strobe-agent:/fixtures:ro + - lab-reports:/reports + command: + - sh + - -ec + - | + /rita import --config /rita-lab/rita-lab.hjson --database lab_traffic --logs /fixtures --rebuild + /rita lab report --config /rita-lab/rita-lab.hjson --lab-config /rita-lab/lab-profile.hjson --window 30m --formats markdown,csv,html --output-dir /reports --overwrite lab_traffic + networks: [lab] + + # Optional profile for a user-supplied, offline PCAP-to-Zeek preprocessing + # stage. It deliberately has no host-network attachment or published ports. + zeek: + image: zeek/zeek:7.0 + profiles: [pcap] + volumes: + - ${LAB_PCAP_DIR:?Set LAB_PCAP_DIR to an offline PCAP directory}:/pcap:ro + - zeek-logs:/logs + command: ["sh", "-ec", "zeek -r /pcap/input.pcap Log::default_logdir=/logs"] + networks: [lab] + +networks: + lab: + internal: true + +volumes: + lab-clickhouse: + lab-reports: + zeek-logs: diff --git a/docs/Configuration.md b/docs/Configuration.md index 4379c2b..27d6f0b 100644 --- a/docs/Configuration.md +++ b/docs/Configuration.md @@ -85,6 +85,12 @@ Inversely, the prevalence modifier also has a score decrease and a decrease thre The Missing Host Header modifier increases the threat score by `missing_host_count_score_increase` if the connection had no host header set. +### RITA-Lab asset and allowlist configuration + +[RITA-Lab](RITALab.md) intentionally reads a separate laboratory HJSON profile for asset tags and report-period allowlist rules. Keep RITA's `filtering.internal_subnets` aligned with the sensor or laboratory source networks: this controls import direction and which records are available to native analysis. + +Do not confuse RITA's `filtering.never_included_domains` with a RITA-Lab allowlist. `never_included_domains` drops matching data during import. RITA-Lab allowlist entries retain the alert and evidence, then record a rule, reason, and visible priority reduction in its generated report. Use import-time exclusion only when retaining evidence is not required. + ### Applying Configuration Changes After making changes to the configuration file, save the file and re-run RITA to apply the changes: diff --git a/docs/RITALab.md b/docs/RITALab.md new file mode 100644 index 0000000..190c18b --- /dev/null +++ b/docs/RITALab.md @@ -0,0 +1,127 @@ +# RITA-Lab: asset-aware, explainable laboratory reporting + +RITA-Lab is a **read-only reporting layer** built into this RITA checkout. It consumes the RITA ClickHouse results after `rita import` has completed. It does not modify RITA detection algorithms, RITA's original score semantics, or the `threat_mixtape` schema. + +```text +Zeek conn.log / dns.log / http.log / ssl.log + -> rita import + -> ClickHouse + RITA native analysis + -> rita lab report + -> asset association, report-period allowlist, aggregation, priority explanation + -> Markdown / CSV / HTML +``` + +## Safety and reproducibility boundary + +The provided lab workflow replays repository-owned Zeek log fixtures only. It does not capture host interfaces, scan systems, contact external targets, or generate attack traffic. The optional Compose `pcap` profile processes a locally supplied PCAP offline and has no host-network attachment. + +The default test inputs are listed in [`lab/fixtures/manifest.hjson`](../lab/fixtures/manifest.hjson). They are controlled RITA test fixtures, not a statement about universal detection coverage or performance. + +## Configure RITA and RITA-Lab separately + +RITA's primary HJSON controls **import-time filtering** and must contain the lab's internal ranges. The sample is [`lab/fixtures/config/rita-lab.hjson`](../lab/fixtures/config/rita-lab.hjson). + +The separate laboratory profile, [`lab/fixtures/config/lab-profile.hjson`](../lab/fixtures/config/lab-profile.hjson), adds: + +- assets (`cidr`, `asset_id`, `asset_type`, `owner`, `importance`, `network_zone`); +- report-period domain allowlist rules; +- secondary-priority weights; +- default aggregation window. + +Asset matching uses the most-specific matching CIDR. An untagged source is retained as `unclassified` rather than discarded. + +### Filtering is not an allowlist + +RITA `filtering.never_included_domains` excludes records **during import**, so later reporting cannot retain their evidence. RITA-Lab's `allowlists.domains` is intentionally different: it retains the alert, annotates the matching pattern/reason, and applies a visible score reduction. This enables auditing the reason for a lower priority. + +Patterns can be exact (`updates.example.test`) or wildcard (`*.updates.example.test`). Domain matching is lower-cased and ignores a trailing dot. + +## Generate a report + +First run a normal import, then choose an explicit report range: + +```bash +rita import --config lab/fixtures/config/rita-lab.hjson \ + --database lab_traffic --logs test_data/dnscat2-ja3-strobe-agent --rebuild + +rita lab report lab_traffic \ + --config lab/fixtures/config/rita-lab.hjson \ + --lab-config lab/fixtures/config/lab-profile.hjson \ + --from 2024-01-01T00:00:00Z --to 2024-01-02T00:00:00Z \ + --window 30m --formats markdown,csv,html \ + --output-dir ./rita-lab-output +``` + +Allowlisted alerts are included by default so the score reduction remains auditable. Add `--exclude-allowlisted` only when an intentionally reduced output view is required. + +If `--from` and `--to` are omitted together, RITA-Lab uses the same recent-range helper as the RITA viewer (the helper caps the lower bound at 24 hours before the latest dataset timestamp). Reports are generated for the UTC half-open interval `[from, to)`. Existing output files are never replaced unless `--overwrite` is passed. + +Use `rita lab evaluate` only for controlled log fixtures. Its `--rebuild` flag is explicit because it destroys and recreates the named database: + +```bash +rita lab evaluate lab_traffic \ + --config lab/fixtures/config/rita-lab.hjson \ + --lab-config lab/fixtures/config/lab-profile.hjson \ + --logs test_data/dnscat2-ja3-strobe-agent \ + --rebuild --output-dir ./rita-lab-output --overwrite +``` + +## Aggregation and evidence + +An alert key is: + +```text +source_ip + destination_domain + detection_type + time_window +``` + +Detection types are `beacon`, `long_connection`, `strobe`, `dns_c2`, and `threat_intel_only`. A native RITA finding can contribute to multiple typed alerts. Missing FQDNs are transparently represented as `ip:`; no domain is fabricated. + +RITA's DNS domain-level result may have `::` as its source. RITA-Lab retains that native result as explicitly unattributed domain evidence; raw `dns.src` records are available as investigation context but are not used to copy the domain-global C2 score onto every querying host. A missing or partial raw match therefore cannot be mistaken for a host-level RITA finding. + +The report additionally shows reproducible DNS investigation features for each source/domain/window: + +- query count and unique encoded labels; +- average and maximum encoded-label length; +- Shannon entropy over encoded-label bytes, in bits per byte; +- queries per minute; +- NXDOMAIN count and ratio. + +These features are clues for investigation, not independent proof of DNS tunneling. + +## Secondary laboratory priority + +Native RITA evidence remains separate from `lab_priority_score`. The default profile uses normalised `0..1` components: + +```text +pre_allowlist_score = + 0.45 * rita_evidence + + 0.20 * asset_importance + + 0.15 * threat_intel + + 0.10 * persistence + + 0.10 * rarity + +lab_priority_score = 100 * clamp(pre_allowlist_score - allowlist_reduction, 0, 1) +``` + +Each output alert includes the raw value, normalised value, configured weight, signed contribution, status, and explanation for every component. Unknown prevalence and unclassified assets are marked `not_available`; they do not receive an invented positive or negative contribution. + +## Report content + +All formats derive from the same report model: + +- **Markdown**: metadata, summary, native evidence, assets, DNS evidence, allowlist decision, and score table; +- **CSV**: structured columns including JSON score breakdown, written with Go's `encoding/csv`; +- **HTML**: static `html/template` output, with all log/configuration values escaped. + +Generated reports record their configuration SHA-256 and scoring version. Evaluation mode records only direct runtime facts (imported record count, aggregated alerts, report duration). It does **not** claim parsing success, accuracy, false-positive rate, recall, or query percentiles unless those metrics are obtained from a completed, labelled evaluation implementation. + +## Docker Compose replay + +Validate and run the isolated replay: + +```bash +docker compose -f docker-compose.lab.yml config +docker compose -f docker-compose.lab.yml up --build --abort-on-container-exit +``` + +Reports are stored in the `lab-reports` Compose volume. The stack has an internal-only network and no published ports. To use the optional offline PCAP stage, set `LAB_PCAP_DIR` to a local directory containing `input.pcap` and invoke `--profile pcap`; inspect and approve the PCAP before running it. diff --git a/lab/aggregate.go b/lab/aggregate.go new file mode 100644 index 0000000..f49f3fb --- /dev/null +++ b/lab/aggregate.go @@ -0,0 +1,143 @@ +package lab + +import ( + "fmt" + "net" + "sort" + "time" +) + +func DetectionTypes(evidence NativeEvidence) []string { + types := make([]string, 0, 5) + if evidence.BeaconThreatScore > 0 || evidence.BeaconScore > 0 { + types = append(types, DetectionBeacon) + } + if evidence.LongConnectionScore > 0 { + types = append(types, DetectionLongConnection) + } + if evidence.StrobeScore > 0 { + types = append(types, DetectionStrobe) + } + if evidence.DNSScore > 0 { + types = append(types, DetectionDNSC2) + } + if evidence.ThreatIntelHit && len(types) == 0 { + types = append(types, DetectionThreatIntelOnly) + } + return types +} + +func AggregateEvidence(evidence []NativeEvidence, events []DNSEvent, cfg *LabConfig, window time.Duration) ([]AggregateAlert, error) { + if cfg == nil { + return nil, fmt.Errorf("lab configuration is nil") + } + if window <= 0 { + return nil, ErrInvalidWindow + } + alerts := make(map[string]*AggregateAlert) + for _, item := range evidence { + for _, detectionType := range DetectionTypes(item) { + if detectionType == DetectionDNSC2 && isUnattributedSource(item.SourceIP) { + // RITA's DNS score is domain-level evidence. A raw DNS query for the + // same domain does not establish that each querying host triggered the + // native score, so retain the result as explicitly unattributed. + item.SourceIP = nil + } + if err := addEvidenceAlert(alerts, item, detectionType, item.LastSeen, cfg, window); err != nil { + return nil, err + } + } + } + + result := make([]AggregateAlert, 0, len(alerts)) + for _, alert := range alerts { + if alert.DetectionType == DetectionDNSC2 && alert.DestinationKind == DestinationDomain && !isUnattributedSource(alert.SourceIP) { + dnsEvents := DNSFeatureEventsForWindow(events, alert.SourceIP.String(), alert.Destination, alert.WindowStart, alert.WindowEnd) + alert.DNSFeatures = CalculateDNSFeatures(dnsEvents, alert.Destination) + alert.DNSQueryCount = alert.DNSFeatures.QueryCount + if alert.DNSFeatures.Available { + alert.FirstSeen = alert.DNSFeatures.FirstSeen + alert.LastSeen = alert.DNSFeatures.LastSeen + } + } + ScoreAlert(alert, cfg.Scoring) + result = append(result, *alert) + } + SortAlerts(result) + return result, nil +} + +func addEvidenceAlert(alerts map[string]*AggregateAlert, item NativeEvidence, detectionType string, timestamp time.Time, cfg *LabConfig, window time.Duration) error { + destination, kind := DestinationFor(item) + if item.SourceIP == nil && detectionType == DetectionDNSC2 { + kind = DestinationUnattributed + } + start, end, err := WindowFor(timestamp, window) + if err != nil { + return err + } + key := fmt.Sprintf("%s\x00%s\x00%s\x00%s", item.SourceIP, destination, detectionType, start) + alert, exists := alerts[key] + if !exists { + alert = &AggregateAlert{ + AlertID: AlertID(item.SourceIP, destination, detectionType, start), + WindowStart: start, + WindowEnd: end, + SourceIP: item.SourceIP, + Destination: destination, + DestinationKind: kind, + DetectionType: detectionType, + FirstSeen: item.FirstSeen, + LastSeen: item.LastSeen, + Asset: MatchAsset(cfg.Assets, item.SourceIP), + Allowlist: MatchAllowlist(cfg.Allowlists.Domains, destination), + } + alerts[key] = alert + } + aggregateInto(alert, item) + return nil +} + +func isUnattributedSource(ip net.IP) bool { + return ip == nil || ip.IsUnspecified() +} + +func aggregateInto(alert *AggregateAlert, item NativeEvidence) { + if alert.FirstSeen.IsZero() || (!item.FirstSeen.IsZero() && item.FirstSeen.Before(alert.FirstSeen)) { + alert.FirstSeen = item.FirstSeen + } + if item.LastSeen.After(alert.LastSeen) { + alert.LastSeen = item.LastSeen + } + alert.ConnectionCount += item.Count + if item.BeaconScore > alert.BeaconScore { + alert.BeaconScore = item.BeaconScore + } + if item.DNSScore > alert.DNSScore { + alert.DNSScore = item.DNSScore + } + alert.ThreatIntelHit = alert.ThreatIntelHit || item.ThreatIntelHit + alert.NativeEvidence = append(alert.NativeEvidence, item) +} + +func SortAlerts(alerts []AggregateAlert) { + sort.Slice(alerts, func(i, j int) bool { + left, right := alerts[i], alerts[j] + if left.LabPriorityScore != right.LabPriorityScore { + return left.LabPriorityScore > right.LabPriorityScore + } + if !left.WindowStart.Equal(right.WindowStart) { + return left.WindowStart.Before(right.WindowStart) + } + if left.SourceIP.String() != right.SourceIP.String() { + return left.SourceIP.String() < right.SourceIP.String() + } + if left.Destination != right.Destination { + return left.Destination < right.Destination + } + if left.DetectionType != right.DetectionType { + return left.DetectionType < right.DetectionType + } + return left.AlertID < right.AlertID + }) +} diff --git a/lab/allowlist.go b/lab/allowlist.go new file mode 100644 index 0000000..6254516 --- /dev/null +++ b/lab/allowlist.go @@ -0,0 +1,29 @@ +package lab + +import ( + "strings" + + "github.com/activecm/rita/v5/util" +) + +func MatchAllowlist(rules []AllowlistRule, domain string) AllowlistMatch { + domain = NormalizeDomain(domain) + for _, rule := range rules { + if !rule.Enabled { + continue + } + pattern := NormalizeAllowlistPattern(rule.Pattern) + if util.ContainsDomain([]string{pattern}, domain) { + return AllowlistMatch{Rule: rule, Matched: true} + } + } + return AllowlistMatch{} +} + +func NormalizeAllowlistPattern(pattern string) string { + pattern = strings.TrimSpace(strings.ToLower(pattern)) + if strings.HasPrefix(pattern, "*.") { + return "*." + NormalizeDomain(strings.TrimPrefix(pattern, "*.")) + } + return NormalizeDomain(pattern) +} diff --git a/lab/assets.go b/lab/assets.go new file mode 100644 index 0000000..70b4033 --- /dev/null +++ b/lab/assets.go @@ -0,0 +1,31 @@ +package lab + +import "net" + +func MatchAsset(assets []Asset, ip net.IP) AssetMatch { + if ip == nil { + return AssetMatch{Asset: unclassifiedAsset()} + } + + best := -1 + var match Asset + for _, asset := range assets { + if asset.network != nil && asset.network.Contains(ip.To16()) && asset.prefix > best { + best = asset.prefix + match = asset + } + } + if best < 0 { + return AssetMatch{Asset: unclassifiedAsset()} + } + return AssetMatch{Asset: match, Matched: true} +} + +func unclassifiedAsset() Asset { + return Asset{ + AssetID: "unclassified", + AssetType: "unclassified", + Owner: "unclassified", + NetworkZone: "unclassified", + } +} diff --git a/lab/config.go b/lab/config.go new file mode 100644 index 0000000..abb9a24 --- /dev/null +++ b/lab/config.go @@ -0,0 +1,98 @@ +package lab + +import ( + "crypto/sha256" + "errors" + "fmt" + "math" + "os" + "strings" + "time" + + "github.com/activecm/rita/v5/util" + "github.com/hjson/hjson-go/v4" +) + +var ( + ErrInvalidSchemaVersion = errors.New("lab schema_version is required") + ErrInvalidAsset = errors.New("invalid lab asset") + ErrInvalidAllowlist = errors.New("invalid lab allowlist rule") + ErrInvalidScoring = errors.New("invalid lab scoring configuration") + ErrInvalidWindow = errors.New("invalid lab reporting window") +) + +func ReadConfig(path string) (*LabConfig, string, error) { + contents, err := os.ReadFile(path) + if err != nil { + return nil, "", err + } + + var cfg LabConfig + if err := hjson.Unmarshal(contents, &cfg); err != nil { + return nil, "", fmt.Errorf("parse lab config: %w", err) + } + if err := cfg.Validate(); err != nil { + return nil, "", err + } + + digest := sha256.Sum256(contents) + return &cfg, fmt.Sprintf("%x", digest), nil +} + +func (cfg *LabConfig) Validate() error { + if strings.TrimSpace(cfg.SchemaVersion) == "" { + return ErrInvalidSchemaVersion + } + window, err := time.ParseDuration(cfg.Reporting.DefaultWindow) + if err != nil || window <= 0 { + if err != nil { + return fmt.Errorf("%w: %w", ErrInvalidWindow, err) + } + return ErrInvalidWindow + } + + assetIDs := make(map[string]struct{}, len(cfg.Assets)) + cidrs := make(map[string]struct{}, len(cfg.Assets)) + for i := range cfg.Assets { + asset := &cfg.Assets[i] + if strings.TrimSpace(asset.AssetID) == "" || strings.TrimSpace(asset.AssetType) == "" || strings.TrimSpace(asset.Owner) == "" || strings.TrimSpace(asset.NetworkZone) == "" || asset.Importance < 0 || asset.Importance > 100 { + return fmt.Errorf("%w at index %d", ErrInvalidAsset, i) + } + if _, ok := assetIDs[asset.AssetID]; ok { + return fmt.Errorf("%w: duplicate asset_id %q", ErrInvalidAsset, asset.AssetID) + } + subnet, err := util.ParseSubnet(asset.CIDR) + if err != nil { + return fmt.Errorf("%w: asset %q CIDR: %w", ErrInvalidAsset, asset.AssetID, err) + } + canonicalCIDR := subnet.ToString() + if _, ok := cidrs[canonicalCIDR]; ok { + return fmt.Errorf("%w: duplicate CIDR %q", ErrInvalidAsset, asset.CIDR) + } + ones, _ := subnet.Mask.Size() + asset.network = subnet.IPNet + asset.prefix = ones + assetIDs[asset.AssetID] = struct{}{} + cidrs[canonicalCIDR] = struct{}{} + } + + for i, rule := range cfg.Allowlists.Domains { + if strings.TrimSpace(rule.Pattern) == "" || strings.TrimSpace(rule.Reason) == "" || rule.ScoreReduction < 0 || rule.ScoreReduction > 1 { + return fmt.Errorf("%w at index %d", ErrInvalidAllowlist, i) + } + } + + s := cfg.Scoring + weights := []float64{s.RITAEvidenceWeight, s.AssetImportanceWeight, s.ThreatIntelWeight, s.PersistenceWeight, s.RarityWeight} + var total float64 + for _, weight := range weights { + if weight < 0 || weight > 1 { + return ErrInvalidScoring + } + total += weight + } + if math.Abs(total-1) > 0.000001 || s.PersistenceCountTarget == 0 || s.PersistenceDurationMins <= 0 { + return ErrInvalidScoring + } + return nil +} diff --git a/lab/dns.go b/lab/dns.go new file mode 100644 index 0000000..36204ad --- /dev/null +++ b/lab/dns.go @@ -0,0 +1,81 @@ +package lab + +import ( + "math" + "strings" + "time" +) + +func CalculateDNSFeatures(events []DNSEvent, domain string) DNSFeatures { + if len(events) == 0 { + return DNSFeatures{} + } + + features := DNSFeatures{Available: true} + domain = NormalizeDomain(domain) + labels := make(map[string]struct{}) + bytes := make(map[byte]uint64) + var byteCount, totalLength, labelCount uint64 + + for _, event := range events { + if domain != "" && NormalizeDomain(event.Domain) != domain { + continue + } + if features.FirstSeen.IsZero() || event.Timestamp.Before(features.FirstSeen) { + features.FirstSeen = event.Timestamp + } + if event.Timestamp.After(features.LastSeen) { + features.LastSeen = event.Timestamp + } + features.QueryCount++ + if strings.EqualFold(event.ResponseCodeName, "NXDOMAIN") { + features.NXDOMAINCount++ + } + + label := EncodedLabel(event.Query, domain) + if label == "" { + continue + } + labels[label] = struct{}{} + length := len(label) + labelCount++ + totalLength += uint64(length) + if length > features.MaximumLabelLength { + features.MaximumLabelLength = length + } + for _, b := range []byte(label) { + bytes[b]++ + byteCount++ + } + } + + features.UniqueEncodedLabels = uint64(len(labels)) + features.NXDOMAINRatio = float64(features.NXDOMAINCount) / float64(features.QueryCount) + if features.LastSeen.After(features.FirstSeen) { + minutes := features.LastSeen.Sub(features.FirstSeen).Minutes() + if minutes > 0 { + features.QueryFrequencyPerMinute = float64(features.QueryCount) / minutes + } + } + if labelCount > 0 { + features.AverageLabelLength = float64(totalLength) / float64(labelCount) + } + if byteCount > 0 { + for _, count := range bytes { + probability := float64(count) / float64(byteCount) + features.LabelEntropyBitsPerByte -= probability * math.Log2(probability) + } + } + return features +} + +func DNSFeatureEventsForWindow(events []DNSEvent, source string, domain string, start, end time.Time) []DNSEvent { + result := make([]DNSEvent, 0) + for _, event := range events { + if event.SourceIP.String() != source || event.Timestamp.Before(start) || !event.Timestamp.Before(end) || NormalizeDomain(event.Domain) != NormalizeDomain(domain) { + continue + } + result = append(result, event) + } + return result +} diff --git a/lab/evaluate.go b/lab/evaluate.go new file mode 100644 index 0000000..947fa8a --- /dev/null +++ b/lab/evaluate.go @@ -0,0 +1,22 @@ +package lab + +import ( + "time" + + "github.com/activecm/rita/v5/importer" +) + +func NewEvaluationResult(importResults importer.ResultCounts, started time.Time, reportDuration time.Duration, aggregatedAlerts int) EvaluationResult { + return EvaluationResult{ + ImportStartedAt: started, + ReportDuration: reportDuration, + ImportedRecords: importResults.Conn + importResults.OpenConn + importResults.HTTP + importResults.OpenHTTP + importResults.DNS + importResults.PDNSRaw + importResults.SSL + importResults.OpenSSL, + AggregatedAlerts: aggregatedAlerts, + Status: "measured", + } +} + +func WithEvaluation(report Report, result EvaluationResult) Report { + report.Evaluation = &result + return report +} diff --git a/lab/fixtures/config/lab-profile.hjson b/lab/fixtures/config/lab-profile.hjson new file mode 100644 index 0000000..74b2a4e --- /dev/null +++ b/lab/fixtures/config/lab-profile.hjson @@ -0,0 +1,47 @@ +{ + schema_version: "v1" + assets: [ + { + cidr: "192.168.0.0/24" + asset_id: "lab-workstations" + asset_type: "workstation" + owner: "lab-operations" + importance: 40 + network_zone: "lab" + } + { + cidr: "192.168.0.15" + asset_id: "lab-critical-server" + asset_type: "server" + owner: "lab-operations" + importance: 90 + network_zone: "restricted" + } + ] + allowlists: { + domains: [ + { + pattern: "*.updates.example.test" + score_reduction: 0.30 + reason: "controlled software-update test endpoint" + enabled: true + } + ] + } + scoring: { + version: "v1" + rita_evidence_weight: 0.45 + asset_importance_weight: 0.20 + threat_intel_weight: 0.15 + persistence_weight: 0.10 + rarity_weight: 0.10 + persistence_count_target: 20 + persistence_duration_minutes: 30 + } + reporting: { + default_window: "30m" + } + dns_features: { + enabled: true + } +} diff --git a/lab/fixtures/config/rita-lab.hjson b/lab/fixtures/config/rita-lab.hjson new file mode 100644 index 0000000..8238e92 --- /dev/null +++ b/lab/fixtures/config/rita-lab.hjson @@ -0,0 +1,12 @@ +{ + // RITA import configuration for repository-owned, offline Zeek log fixtures. + update_check_enabled: false + filtering: { + filter_external_to_internal: false + internal_subnets: ["10.0.0.0/8", "172.16.0.0/12", "192.168.0.0/16", "fd00::/8"] + } + threat_intel: { + // Keep the lab reproducible and offline. Add a local feed only when needed. + online_feeds: [] + } +} diff --git a/lab/fixtures/manifest.hjson b/lab/fixtures/manifest.hjson new file mode 100644 index 0000000..db0a399 --- /dev/null +++ b/lab/fixtures/manifest.hjson @@ -0,0 +1,29 @@ +{ + schema_version: "v1" + fixture_id: "rita-existing-zeek-fixtures" + version: "1" + description: "Offline, repository-owned Zeek log fixtures used for RITA-Lab reproducibility. This manifest intentionally does not assert detection-rate or performance values before execution." + log_sets: [ + { + path: "test_data/dnscat2-ja3-strobe-agent" + purpose: "Existing RITA integration scenario with DNS C2, JA3, and strobe-related evidence." + } + { + path: "test_data/valid_tsv" + purpose: "Valid conn, DNS, HTTP, and SSL import coverage." + } + { + path: "test_data/dns_only" + purpose: "DNS-specific import and feature-query coverage." + } + ] + metrics: { + parse_success_rate: "not_measured" + query_latency: "not_measured" + aggregation_reduction: "not_measured" + false_positive_rate: "not_measured" + detection_coverage: "not_measured" + alert_latency: "not_measured" + allowlist_effect: "not_measured" + } +} diff --git a/lab/lab_test.go b/lab/lab_test.go new file mode 100644 index 0000000..9dee82a --- /dev/null +++ b/lab/lab_test.go @@ -0,0 +1,102 @@ +package lab + +import ( + "net" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +func testConfig(t *testing.T) *LabConfig { + t.Helper() + cfg := &LabConfig{ + SchemaVersion: "v1", + Assets: []Asset{ + {CIDR: "10.0.0.0/8", AssetID: "network", AssetType: "network", Owner: "security", Importance: 20, NetworkZone: "lab"}, + {CIDR: "10.1.2.3", AssetID: "critical-host", AssetType: "server", Owner: "security", Importance: 90, NetworkZone: "restricted"}, + }, + Allowlists: Allowlists{Domains: []AllowlistRule{{Pattern: "*.updates.example.test", ScoreReduction: .3, Reason: "controlled update service", Enabled: true}}}, + Scoring: ScoringConfig{Version: "v1", RITAEvidenceWeight: .45, AssetImportanceWeight: .2, ThreatIntelWeight: .15, PersistenceWeight: .1, RarityWeight: .1, PersistenceCountTarget: 10, PersistenceDurationMins: 30}, + Reporting: ReportingConfig{DefaultWindow: "30m"}, + } + require.NoError(t, cfg.Validate()) + return cfg +} + +func TestMatchAssetUsesLongestPrefix(t *testing.T) { + cfg := testConfig(t) + match := MatchAsset(cfg.Assets, net.ParseIP("10.1.2.3")) + require.True(t, match.Matched) + require.Equal(t, "critical-host", match.Asset.AssetID) + + unmatched := MatchAsset(cfg.Assets, net.ParseIP("192.0.2.1")) + require.False(t, unmatched.Matched) + require.Equal(t, "unclassified", unmatched.Asset.AssetID) +} + +func TestAllowlistNormalizesDomains(t *testing.T) { + cfg := testConfig(t) + match := MatchAllowlist(cfg.Allowlists.Domains, "HOST.UPDATES.EXAMPLE.TEST.") + require.True(t, match.Matched) + require.Equal(t, .3, match.Rule.ScoreReduction) +} + +func TestCalculateDNSFeatures(t *testing.T) { + start := time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC) + features := CalculateDNSFeatures([]DNSEvent{ + {Timestamp: start, Query: "abcd.tunnel.example.test", Domain: "tunnel.example.test", ResponseCodeName: "NXDOMAIN"}, + {Timestamp: start.Add(2 * time.Minute), Query: "efgh.tunnel.example.test", Domain: "tunnel.example.test", ResponseCodeName: "NOERROR"}, + }, "tunnel.example.test") + require.True(t, features.Available) + require.Equal(t, uint64(2), features.QueryCount) + require.Equal(t, uint64(2), features.UniqueEncodedLabels) + require.Equal(t, 4.0, features.AverageLabelLength) + require.Equal(t, 4, features.MaximumLabelLength) + require.Equal(t, .5, features.NXDOMAINRatio) + require.Equal(t, 1.0, features.QueryFrequencyPerMinute) + require.Greater(t, features.LabelEntropyBitsPerByte, 0.0) +} + +func TestAggregateEvidenceScoresAndRetainsAllowlistedAlert(t *testing.T) { + cfg := testConfig(t) + lastSeen := time.Date(2026, 1, 1, 0, 15, 0, 0, time.UTC) + alerts, err := AggregateEvidence([]NativeEvidence{{ + SourceIP: net.ParseIP("10.1.2.3"), FQDN: "host.updates.example.test.", Count: 10, + FirstSeen: lastSeen.Add(-30 * time.Minute), LastSeen: lastSeen, BeaconScore: .8, BeaconThreatScore: .8, + ThreatIntelHit: true, Prevalence: .2, NetworkSize: 20, + }}, nil, cfg, 30*time.Minute) + require.NoError(t, err) + require.Len(t, alerts, 1) + alert := alerts[0] + require.True(t, alert.Allowlist.Matched) + require.Equal(t, "critical-host", alert.Asset.Asset.AssetID) + require.Equal(t, uint64(10), alert.ConnectionCount) + require.Greater(t, alert.PreAllowlistScore, alert.LabPriorityScore/100) + require.Len(t, alert.ScoreBreakdown, 6) +} + +func TestUnattributedDNSUsesRawDNSSource(t *testing.T) { + cfg := testConfig(t) + when := time.Date(2026, 1, 1, 0, 10, 0, 0, time.UTC) + alerts, err := AggregateEvidence([]NativeEvidence{{ + SourceIP: net.ParseIP("::"), FQDN: "tunnel.example.test", DNSScore: .8, LastSeen: when, FirstSeen: when, + }}, []DNSEvent{{ + Timestamp: when, SourceIP: net.ParseIP("10.1.2.3"), Domain: "tunnel.example.test", Query: "encoded.tunnel.example.test", + }}, cfg, 30*time.Minute) + require.NoError(t, err) + require.Len(t, alerts, 1) + require.Nil(t, alerts[0].SourceIP) + require.Equal(t, DestinationUnattributed, alerts[0].DestinationKind) + require.Equal(t, uint64(0), alerts[0].DNSQueryCount) +} + +func TestWindowAndAlertIDAreStable(t *testing.T) { + timestamp := time.Date(2026, 1, 1, 0, 31, 0, 0, time.UTC) + start, end, err := WindowFor(timestamp, 30*time.Minute) + require.NoError(t, err) + require.Equal(t, time.Date(2026, 1, 1, 0, 30, 0, 0, time.UTC), start) + require.Equal(t, start.Add(30*time.Minute), end) + first := AlertID(net.ParseIP("10.0.0.1"), "example.test", DetectionBeacon, start) + require.Equal(t, first, AlertID(net.ParseIP("10.0.0.1"), "example.test", DetectionBeacon, start)) +} diff --git a/lab/model.go b/lab/model.go new file mode 100644 index 0000000..715e4af --- /dev/null +++ b/lab/model.go @@ -0,0 +1,186 @@ +// Package lab adds asset-aware, explainable reporting on top of RITA analysis data. +package lab + +import ( + "net" + "time" +) + +const ( + DetectionBeacon = "beacon" + DetectionLongConnection = "long_connection" + DetectionStrobe = "strobe" + DetectionDNSC2 = "dns_c2" + DetectionThreatIntelOnly = "threat_intel_only" + + DestinationDomain = "domain" + DestinationIPFallback = "ip_fallback" + DestinationUnattributed = "unattributed" +) + +type LabConfig struct { + SchemaVersion string `json:"schema_version"` + Assets []Asset `json:"assets"` + Allowlists Allowlists `json:"allowlists"` + Scoring ScoringConfig `json:"scoring"` + Reporting ReportingConfig `json:"reporting"` + DNSFeatures DNSFeatureConfig `json:"dns_features"` +} + +type Asset struct { + CIDR string `json:"cidr"` + AssetID string `json:"asset_id"` + AssetType string `json:"asset_type"` + Owner string `json:"owner"` + Importance int `json:"importance"` + NetworkZone string `json:"network_zone"` + + network *net.IPNet + prefix int +} + +type Allowlists struct { + Domains []AllowlistRule `json:"domains"` +} + +type AllowlistRule struct { + Pattern string `json:"pattern"` + ScoreReduction float64 `json:"score_reduction"` + Reason string `json:"reason"` + Enabled bool `json:"enabled"` +} + +type ScoringConfig struct { + Version string `json:"version"` + RITAEvidenceWeight float64 `json:"rita_evidence_weight"` + AssetImportanceWeight float64 `json:"asset_importance_weight"` + ThreatIntelWeight float64 `json:"threat_intel_weight"` + PersistenceWeight float64 `json:"persistence_weight"` + RarityWeight float64 `json:"rarity_weight"` + PersistenceCountTarget uint64 `json:"persistence_count_target"` + PersistenceDurationMins float64 `json:"persistence_duration_minutes"` +} + +type ReportingConfig struct { + DefaultWindow string `json:"default_window"` +} + +type DNSFeatureConfig struct { + Enabled bool `json:"enabled"` +} + +type NativeEvidence struct { + Hash string + ImportID string + SourceIP net.IP + DestinationIP net.IP + FQDN string + Count uint64 + LastSeen time.Time + FirstSeen time.Time + BeaconScore float64 + BeaconThreatScore float64 + LongConnectionScore float64 + StrobeScore float64 + DNSScore float64 + ThreatIntelHit bool + ThreatIntelScore float64 + Prevalence float64 + NetworkSize uint64 + BaseScore float64 + TotalModifierScore float64 + NativeFinalScore float64 + Modifiers []Modifier +} + +type Modifier struct { + Name string `json:"name"` + Value string `json:"value"` + Score float64 `json:"score"` +} + +type DNSEvent struct { + Timestamp time.Time + SourceIP net.IP + Domain string + Query string + ResponseCodeName string +} + +type DNSFeatures struct { + QueryCount uint64 `json:"query_count"` + UniqueEncodedLabels uint64 `json:"unique_encoded_labels"` + AverageLabelLength float64 `json:"average_label_length"` + MaximumLabelLength int `json:"maximum_label_length"` + LabelEntropyBitsPerByte float64 `json:"label_entropy_bits_per_byte"` + QueryFrequencyPerMinute float64 `json:"query_frequency_per_minute"` + NXDOMAINCount uint64 `json:"nxdomain_count"` + NXDOMAINRatio float64 `json:"nxdomain_ratio"` + FirstSeen time.Time `json:"first_seen"` + LastSeen time.Time `json:"last_seen"` + Available bool `json:"available"` +} + +type AssetMatch struct { + Asset Asset `json:"asset"` + Matched bool `json:"matched"` +} + +type AllowlistMatch struct { + Rule AllowlistRule `json:"rule"` + Matched bool `json:"matched"` +} + +type ScoreContribution struct { + Name string `json:"name"` + RawValue float64 `json:"raw_value"` + NormalizedValue float64 `json:"normalized_value"` + Weight float64 `json:"weight"` + SignedContribution float64 `json:"signed_contribution"` + Status string `json:"status"` + Reason string `json:"reason"` +} + +type AggregateAlert struct { + AlertID string + WindowStart time.Time + WindowEnd time.Time + SourceIP net.IP + Destination string + DestinationKind string + DetectionType string + FirstSeen time.Time + LastSeen time.Time + ConnectionCount uint64 + DNSQueryCount uint64 + BeaconScore float64 + DNSScore float64 + ThreatIntelHit bool + Asset AssetMatch + Allowlist AllowlistMatch + NativeEvidence []NativeEvidence + DNSFeatures DNSFeatures + ScoreBreakdown []ScoreContribution + PreAllowlistScore float64 + LabPriorityScore float64 +} + +type EvaluationResult struct { + ImportStartedAt time.Time + ReportDuration time.Duration + ImportedRecords uint64 + AggregatedAlerts int + Status string +} + +type Report struct { + Database string + GeneratedAt time.Time + From time.Time + To time.Time + Window time.Duration + ConfigSHA256 string + ScoringVersion string + Alerts []AggregateAlert + Evaluation *EvaluationResult +} diff --git a/lab/normalize.go b/lab/normalize.go new file mode 100644 index 0000000..69a4ba1 --- /dev/null +++ b/lab/normalize.go @@ -0,0 +1,65 @@ +package lab + +import ( + "crypto/sha256" + "fmt" + "net" + "strings" + "time" +) + +func NormalizeDomain(domain string) string { + return strings.TrimSuffix(strings.ToLower(strings.TrimSpace(domain)), ".") +} + +func WindowFor(timestamp time.Time, window time.Duration) (time.Time, time.Time, error) { + if window <= 0 { + return time.Time{}, time.Time{}, ErrInvalidWindow + } + utc := timestamp.UTC() + start := utc.Truncate(window) + return start, start.Add(window), nil +} + +func DestinationFor(evidence NativeEvidence) (string, string) { + if domain := NormalizeDomain(evidence.FQDN); domain != "" { + return domain, DestinationDomain + } + if evidence.DestinationIP != nil { + return "ip:" + evidence.DestinationIP.String(), DestinationIPFallback + } + return "unattributed", DestinationUnattributed +} + +func AlertID(sourceIP net.IP, destination, detectionType string, windowStart time.Time) string { + input := strings.Join([]string{ + sourceIP.String(), + destination, + detectionType, + windowStart.UTC().Format(time.RFC3339Nano), + }, "\x00") + digest := sha256.Sum256([]byte(input)) + return fmt.Sprintf("lab-%x", digest[:12]) +} + +// SignificantDomain is a deliberately conservative fallback for lab-side grouping. +// ClickHouse applies its public-suffix-aware function in queries; this fallback keeps +// Go-side synthetic fixtures deterministic without claiming full PSL semantics. +func SignificantDomain(query string) string { + query = NormalizeDomain(query) + labels := strings.Split(query, ".") + if len(labels) < 3 { + return query + } + return strings.Join(labels[len(labels)-2:], ".") +} + +func EncodedLabel(query, domain string) string { + query = NormalizeDomain(query) + domain = NormalizeDomain(domain) + if query == "" || domain == "" || query == domain || !strings.HasSuffix(query, "."+domain) { + return "" + } + prefix := strings.TrimSuffix(query, "."+domain) + return strings.Split(prefix, ".")[0] +} diff --git a/lab/query.go b/lab/query.go new file mode 100644 index 0000000..47e120e --- /dev/null +++ b/lab/query.go @@ -0,0 +1,289 @@ +package lab + +import ( + "fmt" + "net" + "strings" + "time" + + "github.com/ClickHouse/clickhouse-go/v2" + "github.com/activecm/rita/v5/database" +) + +type evidenceRow struct { + Hash string `ch:"hash"` + ImportID string `ch:"import_id"` + AnalyzedAt time.Time `ch:"analyzed_at"` + Src net.IP `ch:"src"` + Dst net.IP `ch:"dst"` + FQDN string `ch:"fqdn"` + Count uint64 `ch:"count"` + LastSeen time.Time `ch:"last_seen"` + FirstSeen time.Time `ch:"first_seen_historical"` + BeaconScore float64 `ch:"beacon_score"` + BeaconThreatScore float64 `ch:"beacon_threat_score"` + LongConnectionScore float64 `ch:"long_conn_score"` + StrobeScore float64 `ch:"strobe_score"` + DNSScore float64 `ch:"c2_over_dns_score"` + ThreatIntelHit bool `ch:"threat_intel"` + ThreatIntelScore float64 `ch:"threat_intel_score"` + Prevalence float64 `ch:"prevalence"` + NetworkSize uint64 `ch:"network_size"` + BaseScore float64 `ch:"base_score"` + PrevalenceScore float64 `ch:"prevalence_score"` + FirstSeenScore float64 `ch:"first_seen_score"` + ThreatIntelDataSizeScore float64 `ch:"threat_intel_data_size_score"` + MissingHostHeaderScore float64 `ch:"missing_host_header_score"` + DNSDirectConnScore float64 `ch:"c2_over_dns_direct_conn_score"` + NativeFinalScore float64 `ch:"native_final_score"` +} + +type modifierRow struct { + Hash string `ch:"hash"` + ImportID string `ch:"import_id"` + Name string `ch:"modifier_name"` + Value string `ch:"modifier_value"` + Score float64 `ch:"modifier_score"` +} + +type dnsRow struct { + Timestamp time.Time `ch:"ts"` + SourceIP net.IP `ch:"src"` + Domain string `ch:"domain"` + Query string `ch:"query"` + ResponseCodeName string `ch:"response_code_name"` +} + +// QueryNativeEvidence reads the latest baseline snapshot for each RITA connection +// hash in the requested report range. Modifier rows are read separately from that +// exact snapshot, so neither history nor modifiers inflate native evidence counts. +func QueryNativeEvidence(db *database.DB, from, to time.Time) ([]NativeEvidence, error) { + params := clickhouse.Parameters{ + "database": db.GetSelectedDB(), + "from": fmt.Sprint(from.UTC().Unix()), + "to": fmt.Sprint(to.UTC().Unix()), + } + ctx := db.QueryParameters(params) + query := `--sql + WITH latest AS ( + SELECT + hash, + tupleElement(snapshot, 1) AS import_id, + tupleElement(snapshot, 2) AS last_seen, + tupleElement(snapshot, 3) AS analyzed_at + FROM ( + SELECT hash, argMax(tuple(import_id, last_seen, analyzed_at), tuple(last_seen, analyzed_at)) AS snapshot + FROM {database:Identifier}.threat_mixtape + WHERE modifier_name = '' + AND last_seen >= fromUnixTimestamp({from:Int64}) + AND last_seen < fromUnixTimestamp({to:Int64}) + GROUP BY hash + ) + ) + SELECT + t.hash, t.import_id, t.analyzed_at, t.src, t.dst, t.fqdn, t.count, t.last_seen, + t.first_seen_historical, t.beacon_score, t.beacon_threat_score, + t.long_conn_score, t.strobe_score, t.c2_over_dns_score, + t.threat_intel, t.threat_intel_score, t.prevalence, t.network_size, + greatest(t.beacon_threat_score, t.long_conn_score, t.strobe_score, t.c2_over_dns_score, t.threat_intel_score) AS base_score, + t.prevalence_score, t.first_seen_score, t.threat_intel_data_size_score, + t.missing_host_header_score, t.c2_over_dns_direct_conn_score, + greatest(t.beacon_threat_score, t.long_conn_score, t.strobe_score, t.c2_over_dns_score, t.threat_intel_score) + + t.prevalence_score + t.first_seen_score + t.missing_host_header_score + + t.threat_intel_data_size_score + t.c2_over_dns_direct_conn_score AS native_final_score + FROM {database:Identifier}.threat_mixtape AS t + INNER JOIN latest ON t.hash = latest.hash + AND t.import_id = latest.import_id + AND t.last_seen = latest.last_seen + AND t.analyzed_at = latest.analyzed_at + WHERE t.modifier_name = '' + AND t.last_seen >= fromUnixTimestamp({from:Int64}) + AND t.last_seen < fromUnixTimestamp({to:Int64}) + ` + rows, err := db.Conn.Query(ctx, query) + if err != nil { + return nil, fmt.Errorf("query RITA evidence: %w", err) + } + defer rows.Close() + + evidence := make([]NativeEvidence, 0) + byKey := make(map[string]int) + for rows.Next() { + var row evidenceRow + if err := rows.ScanStruct(&row); err != nil { + return nil, fmt.Errorf("scan RITA evidence: %w", err) + } + item := NativeEvidence{ + Hash: row.Hash, ImportID: row.ImportID, SourceIP: row.Src, DestinationIP: row.Dst, FQDN: row.FQDN, + Count: row.Count, LastSeen: row.LastSeen, FirstSeen: row.FirstSeen, BeaconScore: row.BeaconScore, + BeaconThreatScore: row.BeaconThreatScore, LongConnectionScore: row.LongConnectionScore, StrobeScore: row.StrobeScore, + DNSScore: row.DNSScore, ThreatIntelHit: row.ThreatIntelHit, ThreatIntelScore: row.ThreatIntelScore, + Prevalence: row.Prevalence, NetworkSize: row.NetworkSize, BaseScore: row.BaseScore, + NativeFinalScore: row.NativeFinalScore, + } + if item.FirstSeen.IsZero() { + item.FirstSeen = item.LastSeen + } + byKey[evidenceKey(row.Hash, row.ImportID, row.LastSeen, row.AnalyzedAt)] = len(evidence) + evidence = append(evidence, item) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("iterate RITA evidence: %w", err) + } + + if err := queryModifiers(db, from, to, byKey, evidence); err != nil { + return nil, err + } + return evidence, nil +} + +func evidenceKey(hash, importID string, lastSeen, analyzedAt time.Time) string { + return strings.Join([]string{hash, importID, lastSeen.UTC().Format(time.RFC3339Nano), analyzedAt.UTC().Format(time.RFC3339Nano)}, "\x00") +} + +func queryModifiers(db *database.DB, from, to time.Time, indexes map[string]int, evidence []NativeEvidence) error { + if len(indexes) == 0 { + return nil + } + params := clickhouse.Parameters{ + "database": db.GetSelectedDB(), + "from": fmt.Sprint(from.UTC().Unix()), + "to": fmt.Sprint(to.UTC().Unix()), + } + ctx := db.QueryParameters(params) + query := `--sql + WITH latest AS ( + SELECT + hash, + tupleElement(snapshot, 1) AS import_id, + tupleElement(snapshot, 2) AS last_seen, + tupleElement(snapshot, 3) AS analyzed_at + FROM ( + SELECT hash, argMax(tuple(import_id, last_seen, analyzed_at), tuple(last_seen, analyzed_at)) AS snapshot + FROM {database:Identifier}.threat_mixtape + WHERE modifier_name = '' + AND last_seen >= fromUnixTimestamp({from:Int64}) + AND last_seen < fromUnixTimestamp({to:Int64}) + GROUP BY hash + ) + ) + SELECT t.hash, t.import_id, t.modifier_name, t.modifier_value, t.modifier_score + FROM {database:Identifier}.threat_mixtape AS t + INNER JOIN latest ON t.hash = latest.hash + AND t.import_id = latest.import_id + AND t.last_seen = latest.last_seen + AND t.analyzed_at = latest.analyzed_at + WHERE t.modifier_name != '' + AND t.last_seen >= fromUnixTimestamp({from:Int64}) + AND t.last_seen < fromUnixTimestamp({to:Int64}) + ` + rows, err := db.Conn.Query(ctx, query) + if err != nil { + return fmt.Errorf("query RITA modifiers: %w", err) + } + defer rows.Close() + for rows.Next() { + var row modifierRow + if err := rows.ScanStruct(&row); err != nil { + return fmt.Errorf("scan RITA modifier: %w", err) + } + var index int + var ok bool + // The SQL snapshot join leaves one baseline snapshot per hash/import pair; + // modifier rows carry the same pair but not the baseline timestamps. + for candidateKey, candidateIndex := range indexes { + if strings.HasPrefix(candidateKey, row.Hash+"\x00"+row.ImportID+"\x00") { + index, ok = candidateIndex, true + break + } + } + if !ok { + continue + } + evidence[index].Modifiers = append(evidence[index].Modifiers, Modifier{Name: row.Name, Value: row.Value, Score: row.Score}) + evidence[index].TotalModifierScore += row.Score + evidence[index].NativeFinalScore += row.Score + } + if err := rows.Err(); err != nil { + return fmt.Errorf("iterate RITA modifiers: %w", err) + } + return nil +} + +func QueryDNSEvents(db *database.DB, from, to time.Time, domains []string) ([]DNSEvent, error) { + domains = normalizedDomains(domains) + if len(domains) == 0 { + return nil, nil + } + params := clickhouse.Parameters{ + "database": db.GetSelectedDB(), + "from": fmt.Sprint(from.UTC().Unix()), + "to": fmt.Sprint(to.UTC().Unix()), + "domains": clickHouseStringArray(domains), + } + ctx := db.QueryParameters(params) + query := `--sql + SELECT ts, src, lowerUTF8(trimRight(cutToFirstSignificantSubdomain(query), '.')) AS domain, query, response_code_name + FROM {database:Identifier}.dns + WHERE ts >= fromUnixTimestamp({from:Int64}) + AND ts < fromUnixTimestamp({to:Int64}) + AND lowerUTF8(trimRight(cutToFirstSignificantSubdomain(query), '.')) IN {domains:Array(String)} + ` + rows, err := db.Conn.Query(ctx, query) + if err != nil { + return nil, fmt.Errorf("query raw DNS evidence: %w", err) + } + defer rows.Close() + events := make([]DNSEvent, 0) + for rows.Next() { + var row dnsRow + if err := rows.ScanStruct(&row); err != nil { + return nil, fmt.Errorf("scan raw DNS evidence: %w", err) + } + events = append(events, DNSEvent{Timestamp: row.Timestamp, SourceIP: row.SourceIP, Domain: NormalizeDomain(row.Domain), Query: row.Query, ResponseCodeName: row.ResponseCodeName}) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("iterate raw DNS evidence: %w", err) + } + return events, nil +} + +func normalizedDomains(domains []string) []string { + seen := make(map[string]struct{}, len(domains)) + result := make([]string, 0, len(domains)) + for _, domain := range domains { + domain = NormalizeDomain(domain) + if domain == "" { + continue + } + if _, ok := seen[domain]; ok { + continue + } + seen[domain] = struct{}{} + result = append(result, domain) + } + return result +} + +// clickHouseStringArray returns a ClickHouse Array(String) literal for the +// driver's string-only named-parameter API. Values originate from database +// results but remain escaped rather than interpolated into the SQL text. +func clickHouseStringArray(values []string) string { + quoted := make([]string, 0, len(values)) + for _, value := range values { + value = strings.ReplaceAll(value, "\\", "\\\\") + value = strings.ReplaceAll(value, "'", "\\'") + quoted = append(quoted, "'"+value+"'") + } + return "[" + strings.Join(quoted, ",") + "]" +} + +func DNSEvidenceDomains(evidence []NativeEvidence) []string { + domains := make([]string, 0) + for _, item := range evidence { + if item.DNSScore > 0 && NormalizeDomain(item.FQDN) != "" { + domains = append(domains, item.FQDN) + } + } + return normalizedDomains(domains) +} diff --git a/lab/report.go b/lab/report.go new file mode 100644 index 0000000..dc21a12 --- /dev/null +++ b/lab/report.go @@ -0,0 +1,136 @@ +package lab + +import ( + "bytes" + "encoding/csv" + "encoding/json" + "fmt" + "html/template" + "io" + "strconv" + "strings" + "time" +) + +func NewReport(database string, from, to time.Time, window time.Duration, configSHA256, scoringVersion string, alerts []AggregateAlert) Report { + SortAlerts(alerts) + return Report{ + Database: database, GeneratedAt: time.Now().UTC(), From: from.UTC(), To: to.UTC(), Window: window, + ConfigSHA256: configSHA256, ScoringVersion: scoringVersion, Alerts: alerts, + } +} + +func WriteMarkdown(writer io.Writer, report Report) error { + if _, err := fmt.Fprintf(writer, "# RITA-Lab Threat Hunting Report\n\n"); err != nil { + return err + } + if _, err := fmt.Fprintf(writer, "- Generated (UTC): %s\n- Database: `%s`\n- Data range: `%s` to `%s` (exclusive)\n- Aggregation window: `%s`\n- Lab configuration SHA-256: `%s`\n- Scoring version: `%s`\n", report.GeneratedAt.Format(time.RFC3339), report.Database, report.From.Format(time.RFC3339), report.To.Format(time.RFC3339), report.Window, report.ConfigSHA256, report.ScoringVersion); err != nil { + return err + } + if report.Evaluation != nil { + if _, err := fmt.Fprintf(writer, "- Evaluation: %s; imported records: %d; aggregated alerts: %d; report duration: %s\n", report.Evaluation.Status, report.Evaluation.ImportedRecords, report.Evaluation.AggregatedAlerts, report.Evaluation.ReportDuration.Round(time.Millisecond)); err != nil { + return err + } + } + if _, err := fmt.Fprintln(writer, "\nThis report preserves RITA's native evidence and adds a laboratory triage priority. The laboratory priority does not replace a RITA detection conclusion."); err != nil { + return err + } + if _, err := fmt.Fprintln(writer, "\n## Alert summary\n\n| Priority | Source | Destination | Type | First seen | Last seen | Allowlisted |\n|---:|---|---|---|---|---|---|"); err != nil { + return err + } + for _, alert := range report.Alerts { + if _, err := fmt.Fprintf(writer, "| %.2f | %s | %s | %s | %s | %s | %t |\n", alert.LabPriorityScore, alert.SourceIP, alert.Destination, alert.DetectionType, alert.FirstSeen.UTC().Format(time.RFC3339), alert.LastSeen.UTC().Format(time.RFC3339), alert.Allowlist.Matched); err != nil { + return err + } + } + for _, alert := range report.Alerts { + if _, err := fmt.Fprintf(writer, "\n## %s — %s → %s\n\n", alert.AlertID, alert.SourceIP, alert.Destination); err != nil { + return err + } + if _, err := fmt.Fprintf(writer, "- Detection type: `%s`\n- Window: `%s` to `%s` (exclusive)\n- Native events: %d; connection count: %d; DNS query count: %d\n- Asset: `%s` (%s, importance %d, zone %s)\n- Lab priority: **%.2f/100** (pre-allowlist %.4f)\n", alert.DetectionType, alert.WindowStart.Format(time.RFC3339), alert.WindowEnd.Format(time.RFC3339), len(alert.NativeEvidence), alert.ConnectionCount, alert.DNSQueryCount, alert.Asset.Asset.AssetID, alert.Asset.Asset.AssetType, alert.Asset.Asset.Importance, alert.Asset.Asset.NetworkZone, alert.LabPriorityScore, alert.PreAllowlistScore); err != nil { + return err + } + if alert.Allowlist.Matched { + if _, err := fmt.Fprintf(writer, "- Allowlist: matched `%s`; reduction %.4f; reason: %s\n", alert.Allowlist.Rule.Pattern, alert.Allowlist.Rule.ScoreReduction, alert.Allowlist.Rule.Reason); err != nil { + return err + } + } + if alert.DNSFeatures.Available { + if _, err := fmt.Fprintf(writer, "- DNS evidence: %d queries, %d unique labels, average/max label length %.2f/%d, entropy %.4f bits/byte, NXDOMAIN ratio %.4f\n", alert.DNSFeatures.QueryCount, alert.DNSFeatures.UniqueEncodedLabels, alert.DNSFeatures.AverageLabelLength, alert.DNSFeatures.MaximumLabelLength, alert.DNSFeatures.LabelEntropyBitsPerByte, alert.DNSFeatures.NXDOMAINRatio); err != nil { + return err + } + } + if _, err := fmt.Fprintln(writer, "\n### Score explanation\n\n| Component | Raw | Normalized | Weight | Contribution | Status | Reason |\n|---|---:|---:|---:|---:|---|---|"); err != nil { + return err + } + for _, score := range alert.ScoreBreakdown { + if _, err := fmt.Fprintf(writer, "| %s | %.4f | %.4f | %.4f | %.4f | %s | %s |\n", score.Name, score.RawValue, score.NormalizedValue, score.Weight, score.SignedContribution, score.Status, score.Reason); err != nil { + return err + } + } + } + _, err := fmt.Fprintln(writer, "\n## Interpretation notes\n\nDNS entropy, label length, frequency, and NXDOMAIN ratio are investigation features. They do not independently prove DNS tunneling. Metrics are only produced by `rita lab evaluate` from actual fixture runs; absent metrics are reported as `not_measured`.") + return err +} + +func WriteCSV(writer io.Writer, report Report) error { + csvWriter := csv.NewWriter(writer) + defer csvWriter.Flush() + if err := csvWriter.Write([]string{"alert_id", "window_start", "window_end", "source_ip", "destination_domain", "destination_kind", "detection_type", "first_seen", "last_seen", "connection_count", "dns_query_count", "beacon_score", "dns_score", "threat_intel_hit", "asset_id", "asset_type", "owner", "asset_importance", "network_zone", "allowlisted", "allowlist_rule", "allowlist_reason", "allowlist_reduction", "lab_priority_score", "score_breakdown_json"}); err != nil { + return err + } + for _, alert := range report.Alerts { + breakdown, err := json.Marshal(alert.ScoreBreakdown) + if err != nil { + return err + } + row := []string{ + alert.AlertID, alert.WindowStart.Format(time.RFC3339), alert.WindowEnd.Format(time.RFC3339), alert.SourceIP.String(), alert.Destination, alert.DestinationKind, alert.DetectionType, + alert.FirstSeen.Format(time.RFC3339), alert.LastSeen.Format(time.RFC3339), strconv.FormatUint(alert.ConnectionCount, 10), strconv.FormatUint(alert.DNSQueryCount, 10), + strconv.FormatFloat(alert.BeaconScore, 'f', -1, 64), strconv.FormatFloat(alert.DNSScore, 'f', -1, 64), strconv.FormatBool(alert.ThreatIntelHit), + alert.Asset.Asset.AssetID, alert.Asset.Asset.AssetType, alert.Asset.Asset.Owner, strconv.Itoa(alert.Asset.Asset.Importance), alert.Asset.Asset.NetworkZone, + strconv.FormatBool(alert.Allowlist.Matched), alert.Allowlist.Rule.Pattern, alert.Allowlist.Rule.Reason, strconv.FormatFloat(alert.Allowlist.Rule.ScoreReduction, 'f', -1, 64), strconv.FormatFloat(alert.LabPriorityScore, 'f', -1, 64), string(breakdown), + } + if err := csvWriter.Write(row); err != nil { + return err + } + } + return csvWriter.Error() +} + +var htmlReportTemplate = template.Must(template.New("report").Parse(` +RITA-Lab Threat Hunting Report + +

RITA-Lab Threat Hunting Report

Laboratory priority is triage context, not a replacement for RITA native detection conclusions.

+
  • Database: {{.Database}}
  • Generated (UTC): {{.GeneratedAt}}
  • Data range: {{.From}} to {{.To}} (exclusive)
  • Window: {{.Window}}
  • Configuration SHA-256: {{.ConfigSHA256}}
  • Scoring version: {{.ScoringVersion}}
    • {{if .Evaluation}}
  • Evaluation: {{.Evaluation.Status}}; imported records: {{.Evaluation.ImportedRecords}}; aggregated alerts: {{.Evaluation.AggregatedAlerts}}; report duration: {{.Evaluation.ReportDuration}}
    • {{end}}
+

Alerts

{{range .Alerts}}{{end}}
PrioritySourceDestinationTypeAssetAllowlist
{{printf "%.2f" .LabPriorityScore}}{{.SourceIP}}{{.Destination}}{{.DetectionType}}{{.Asset.Asset.AssetID}}{{if .Allowlist.Matched}}{{.Allowlist.Rule.Pattern}}{{end}}
+{{range .Alerts}}

{{.AlertID}}

{{.DetectionType}}: {{.SourceIP}} → {{.Destination}}; {{.FirstSeen}}–{{.LastSeen}}; native events {{len .NativeEvidence}}.

+

Asset: {{.Asset.Asset.AssetID}} (importance {{.Asset.Asset.Importance}}). {{if .Allowlist.Matched}}Allowlist: {{.Allowlist.Rule.Pattern}}; reduction {{printf "%.4f" .Allowlist.Rule.ScoreReduction}}; reason: {{.Allowlist.Rule.Reason}}.{{end}}

+{{if .DNSFeatures.Available}}

DNS: {{.DNSFeatures.QueryCount}} queries; {{.DNSFeatures.UniqueEncodedLabels}} unique labels; entropy {{printf "%.4f" .DNSFeatures.LabelEntropyBitsPerByte}} bits/byte; NXDOMAIN ratio {{printf "%.4f" .DNSFeatures.NXDOMAINRatio}}.

{{end}} +{{range .ScoreBreakdown}}{{end}}
ComponentRawNormalizedWeightContributionStatusReason
{{.Name}}{{printf "%.4f" .RawValue}}{{printf "%.4f" .NormalizedValue}}{{printf "%.4f" .Weight}}{{printf "%.4f" .SignedContribution}}{{.Status}}{{.Reason}}
{{end}} +`)) + +func WriteHTML(writer io.Writer, report Report) error { + return htmlReportTemplate.Execute(writer, report) +} + +func RenderMarkdown(report Report) (string, error) { + var output bytes.Buffer + if err := WriteMarkdown(&output, report); err != nil { + return "", err + } + return output.String(), nil +} + +func ReportFileExtension(format string) (string, bool) { + switch strings.ToLower(strings.TrimSpace(format)) { + case "markdown", "md": + return "md", true + case "csv": + return "csv", true + case "html": + return "html", true + default: + return "", false + } +} diff --git a/lab/report_test.go b/lab/report_test.go new file mode 100644 index 0000000..cf29908 --- /dev/null +++ b/lab/report_test.go @@ -0,0 +1,44 @@ +package lab + +import ( + "bytes" + "net" + "strings" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +func TestReportFormatsEscapeStructuredContent(t *testing.T) { + cfg := testConfig(t) + when := time.Date(2026, 1, 1, 0, 15, 0, 0, time.UTC) + alerts, err := AggregateEvidence([]NativeEvidence{{ + SourceIP: net.ParseIP("10.1.2.3"), FQDN: "host.updates.example.test", Count: 3, + FirstSeen: when.Add(-time.Minute), LastSeen: when, BeaconScore: .8, BeaconThreatScore: .8, NetworkSize: 1, + }}, nil, cfg, 30*time.Minute) + require.NoError(t, err) + alerts[0].Allowlist.Rule.Reason = "owner, \"quoted\"\nnew line " + report := NewReport("lab_traffic", when.Add(-time.Hour), when.Add(time.Hour), 30*time.Minute, "abc", "v1", alerts) + + var csvOutput bytes.Buffer + require.NoError(t, WriteCSV(&csvOutput, report)) + require.Contains(t, csvOutput.String(), "\"owner, \"\"quoted\"\"\nnew line \"") + + var htmlOutput bytes.Buffer + require.NoError(t, WriteHTML(&htmlOutput, report)) + require.NotContains(t, htmlOutput.String(), "") + require.Contains(t, htmlOutput.String(), "<script>") + + markdown, err := RenderMarkdown(report) + require.NoError(t, err) + require.True(t, strings.Contains(markdown, "RITA-Lab Threat Hunting Report")) +} + +func TestParseFormats(t *testing.T) { + formats, err := ParseFormats(" markdown,CSV,html ") + require.NoError(t, err) + require.Equal(t, []string{"markdown", "csv", "html"}, formats) + _, err = ParseFormats("pdf") + require.Error(t, err) +} diff --git a/lab/run.go b/lab/run.go new file mode 100644 index 0000000..5d8cd45 --- /dev/null +++ b/lab/run.go @@ -0,0 +1,129 @@ +package lab + +import ( + "fmt" + "io" + "os" + "path/filepath" + "strings" + "time" + + "github.com/activecm/rita/v5/database" +) + +func BuildReport(db *database.DB, cfg *LabConfig, configSHA256 string, from, to time.Time, window time.Duration) (Report, error) { + evidence, err := QueryNativeEvidence(db, from, to) + if err != nil { + return Report{}, err + } + var dnsEvents []DNSEvent + if cfg.DNSFeatures.Enabled { + dnsEvents, err = QueryDNSEvents(db, from, to, DNSEvidenceDomains(evidence)) + if err != nil { + return Report{}, err + } + } + alerts, err := AggregateEvidence(evidence, dnsEvents, cfg, window) + if err != nil { + return Report{}, err + } + return NewReport(db.GetSelectedDB(), from, to, window, configSHA256, cfg.Scoring.Version, alerts), nil +} + +func FilterReport(report Report, minimumScore float64, includeAllowlisted bool) Report { + alerts := make([]AggregateAlert, 0, len(report.Alerts)) + for _, alert := range report.Alerts { + if alert.LabPriorityScore < minimumScore || (!includeAllowlisted && alert.Allowlist.Matched) { + continue + } + alerts = append(alerts, alert) + } + report.Alerts = alerts + return report +} + +func WriteReportFiles(report Report, outputDir string, formats []string, overwrite bool) ([]string, error) { + if len(formats) == 0 { + return nil, fmt.Errorf("at least one report format is required") + } + + paths := make([]string, 0, len(formats)) + seen := make(map[string]struct{}) + for _, format := range formats { + extension, ok := ReportFileExtension(format) + if !ok { + return nil, fmt.Errorf("unsupported report format %q", format) + } + if _, duplicate := seen[extension]; duplicate { + continue + } + seen[extension] = struct{}{} + paths = append(paths, filepath.Join(outputDir, "rita-lab-report."+extension)) + } + if len(paths) == 0 { + return nil, fmt.Errorf("at least one report format is required") + } + for _, path := range paths { + if overwrite { + continue + } + if _, err := os.Stat(path); err == nil { + return nil, fmt.Errorf("refusing to overwrite %s; pass --overwrite to replace it", path) + } else if !os.IsNotExist(err) { + return nil, err + } + } + if err := os.MkdirAll(outputDir, 0o755); err != nil { + return nil, err + } + + type renderFunc func(io.Writer) error + renders := make([]renderFunc, len(paths)) + for i, path := range paths { + switch filepath.Ext(path) { + case ".md": + renders[i] = func(writer io.Writer) error { return WriteMarkdown(writer, report) } + case ".csv": + renders[i] = func(writer io.Writer) error { return WriteCSV(writer, report) } + case ".html": + renders[i] = func(writer io.Writer) error { return WriteHTML(writer, report) } + } + } + for i, path := range paths { + tmp, err := os.CreateTemp(outputDir, ".rita-lab-report-*") + if err != nil { + return nil, err + } + tmpName := tmp.Name() + if err := renders[i](tmp); err != nil { + _ = tmp.Close() + _ = os.Remove(tmpName) + return nil, err + } + if err := tmp.Close(); err != nil { + _ = os.Remove(tmpName) + return nil, err + } + if err := os.Rename(tmpName, path); err != nil { + _ = os.Remove(tmpName) + return nil, err + } + } + return paths, nil +} + +func ParseFormats(value string) ([]string, error) { + formats := strings.Split(value, ",") + result := make([]string, 0, len(formats)) + for _, format := range formats { + format = strings.TrimSpace(strings.ToLower(format)) + if _, ok := ReportFileExtension(format); !ok { + return nil, fmt.Errorf("unsupported report format %q", format) + } + result = append(result, format) + } + if len(result) == 0 || (len(result) == 1 && result[0] == "") { + return nil, fmt.Errorf("at least one report format is required") + } + return result, nil +} diff --git a/lab/scoring.go b/lab/scoring.go new file mode 100644 index 0000000..bd24d25 --- /dev/null +++ b/lab/scoring.go @@ -0,0 +1,133 @@ +package lab + +import ( + "fmt" + "math" + "time" +) + +func ScoreAlert(alert *AggregateAlert, cfg ScoringConfig) { + evidence, available := evidenceScore(*alert) + contributions := []ScoreContribution{ + contribution("rita_evidence", evidence, cfg.RITAEvidenceWeight, available, "native RITA evidence for this detection type"), + assetContribution(alert.Asset, cfg.AssetImportanceWeight), + contribution("threat_intel", boolScore(alert.ThreatIntelHit), cfg.ThreatIntelWeight, true, "RITA threat-intel hit state"), + persistenceContribution(*alert, cfg), + rarityContribution(*alert, cfg.RarityWeight), + } + + preAllowlist := 0.0 + for _, item := range contributions { + preAllowlist += item.SignedContribution + } + if alert.Allowlist.Matched { + reduction := clamp(alert.Allowlist.Rule.ScoreReduction) + contributions = append(contributions, ScoreContribution{ + Name: "allowlist_reduction", + RawValue: reduction, + NormalizedValue: reduction, + SignedContribution: -reduction, + Status: "allowlist_reduction", + Reason: alert.Allowlist.Rule.Reason, + }) + } + + alert.PreAllowlistScore = clamp(preAllowlist) + if alert.Allowlist.Matched { + preAllowlist -= clamp(alert.Allowlist.Rule.ScoreReduction) + } + alert.LabPriorityScore = 100 * clamp(preAllowlist) + alert.ScoreBreakdown = contributions +} + +func evidenceScore(alert AggregateAlert) (float64, bool) { + switch alert.DetectionType { + case DetectionBeacon: + return clamp(maxNative(alert.NativeEvidence, func(e NativeEvidence) float64 { return max(e.BeaconThreatScore, e.BeaconScore) })), true + case DetectionLongConnection: + return clamp(maxNative(alert.NativeEvidence, func(e NativeEvidence) float64 { return e.LongConnectionScore })), true + case DetectionStrobe: + return clamp(maxNative(alert.NativeEvidence, func(e NativeEvidence) float64 { return e.StrobeScore })), true + case DetectionDNSC2: + return clamp(maxNative(alert.NativeEvidence, func(e NativeEvidence) float64 { return e.DNSScore })), true + case DetectionThreatIntelOnly: + return clamp(maxNative(alert.NativeEvidence, func(e NativeEvidence) float64 { return e.ThreatIntelScore })), true + default: + return 0, false + } +} + +func maxNative(evidence []NativeEvidence, value func(NativeEvidence) float64) float64 { + var best float64 + for _, item := range evidence { + best = max(best, value(item)) + } + return best +} + +func contribution(name string, raw, weight float64, available bool, reason string) ScoreContribution { + status := "applied" + if !available { + status = "not_available" + } + normalized := clamp(raw) + return ScoreContribution{ + Name: name, + RawValue: raw, + NormalizedValue: normalized, + Weight: weight, + SignedContribution: normalized * weight, + Status: status, + Reason: reason, + } +} + +func assetContribution(asset AssetMatch, weight float64) ScoreContribution { + if !asset.Matched { + return contribution("asset_importance", 0, weight, false, "source asset is not tagged") + } + return contribution("asset_importance", float64(asset.Asset.Importance)/100, weight, true, "matched asset "+asset.Asset.AssetID) +} + +func persistenceContribution(alert AggregateAlert, cfg ScoringConfig) ScoreContribution { + count := alert.ConnectionCount + span := alert.LastSeen.Sub(alert.FirstSeen) + if alert.DetectionType == DetectionDNSC2 && alert.DNSFeatures.Available { + count = alert.DNSFeatures.QueryCount + span = alert.DNSFeatures.LastSeen.Sub(alert.DNSFeatures.FirstSeen) + } + countComponent := clamp(float64(count) / float64(cfg.PersistenceCountTarget)) + durationComponent := clamp(span.Minutes() / cfg.PersistenceDurationMins) + return contribution("persistence", (countComponent+durationComponent)/2, cfg.PersistenceWeight, true, fmt.Sprintf("%d events across %s", count, span.Round(time.Second))) +} + +func rarityContribution(alert AggregateAlert, weight float64) ScoreContribution { + var prevalence float64 + available := false + for _, evidence := range alert.NativeEvidence { + if evidence.NetworkSize > 0 && evidence.Prevalence >= 0 { + prevalence = evidence.Prevalence + available = true + break + } + } + if !available { + return contribution("rarity", 0, weight, false, "RITA prevalence is unavailable") + } + return contribution("rarity", 1-clamp(prevalence), weight, true, "inverse RITA prevalence") +} + +func boolScore(value bool) float64 { + if value { + return 1 + } + return 0 +} + +func clamp(value float64) float64 { + return math.Max(0, math.Min(1, value)) +} + +func max(left, right float64) float64 { + return math.Max(left, right) +}