package main import ( "bytes" "context" "encoding/json" "errors" "flag" "fmt" "io" "net" "net/http" "net/netip" "os" "os/exec" "path/filepath" "sort" "sync/atomic" "time" ) const ( defaultConfigPath = "/etc/oci-l4-relay-smtp-evidence.json" defaultNftPath = "/usr/sbin/nft" defaultSnapshotPath = "/var/lib/oci-l4-relay/smtp-evidence/current.json" defaultTCPath = "/usr/sbin/tc" defaultTelemetryPath = "/run/oci-l4-relay-telemetry/current.json" nftablesFamily = "inet" nftablesTable = "oci_l4_relay" reason = "syn-rate" ) var evidenceSets = []evidenceSet{ {Service: "web", Name: "syn_quarantine_v4"}, {Service: "smtp", Name: "smtp_syn_quarantine_v4"}, } var startConntrackObserver = observeConntrack var dropMetricSpecs = []dropMetricSpec{ {Name: "web_allow_denied", Service: "web", Reason: "admission"}, {Name: "web_syn_source_dropped", Service: "web", Reason: "source_rate"}, {Name: "web_syn_quarantine_dropped", Service: "web", Reason: "quarantine"}, {Name: "web_syn_aggregate_dropped", Service: "web", Reason: "aggregate_syn"}, {Name: "web_packet_aggregate_dropped", Service: "web", Reason: "aggregate_packet"}, {Name: "smtp_syn_source_dropped", Service: "smtp", Reason: "source_rate"}, {Name: "smtp_syn_quarantine_dropped", Service: "smtp", Reason: "quarantine"}, {Name: "smtp_syn_aggregate_dropped", Service: "smtp", Reason: "aggregate_syn"}, {Name: "smtp_packet_aggregate_dropped", Service: "smtp", Reason: "aggregate_packet"}, {Name: "smtp_denied", Service: "smtp", Reason: "deny_policy", MailOnly: true}, {Name: "auth_syn_source_dropped", Service: "authenticated_mail", Reason: "source_rate"}, {Name: "auth_syn_quarantine_dropped", Service: "authenticated_mail", Reason: "quarantine"}, {Name: "auth_syn_aggregate_dropped", Service: "authenticated_mail", Reason: "aggregate_syn"}, {Name: "auth_packet_aggregate_dropped", Service: "authenticated_mail", Reason: "aggregate_packet"}, } type relayConfig struct { ConntrackEmissionBurst int `json:"conntrack_emission_burst"` ConntrackEmissionRatePerSecond int `json:"conntrack_emission_rate_per_second"` ConntrackEventBuffer int `json:"conntrack_event_buffer"` ConntrackSampleDenominator uint64 `json:"conntrack_sample_denominator"` ConntrackServices []conntrackService `json:"conntrack_services"` MaximumRecords uint64 `json:"maximum_records"` MetricsListenAddress string `json:"metrics_listen_address"` ObservationWindowSeconds int `json:"observation_window_seconds"` SnapshotIntervalSeconds int `json:"snapshot_interval_seconds"` } type conntrackService struct { Name string `json:"name"` Ports []uint16 `json:"ports"` TargetAddress netip.Addr `json:"target_address"` } type evidenceSet struct { Name string Service string } type evidenceRecord struct { Address string `json:"address"` Count uint64 `json:"count"` ExpiresAt time.Time `json:"expires_at"` } type snapshot struct { SchemaVersion int `json:"schema_version"` GeneratedAt time.Time `json:"generated_at"` Reason string `json:"reason"` ObservationWindowSeconds int `json:"observation_window_seconds"` Records []evidenceRecord `json:"records"` } type serviceMetrics struct { Records []evidenceRecord Service string Total uint64 } type dropMetric struct { Packets uint64 Reason string Service string } type dropMetricSpec struct { MailOnly bool Name string Reason string Service string } type metricsSnapshot struct { Drops []dropMetric GeneratedAt time.Time Health relayHealth QdiscDrops []qdiscMetric Services []serviceMetrics TelemetryAt time.Time } type qdiscMetric struct { Channel string Packets uint64 } type relayHealth struct { ConntrackCount int64 `json:"conntrack_count"` ConntrackMaximum int64 `json:"conntrack_maximum"` HandshakeAgeSeconds int64 `json:"handshake_age_seconds"` NftablesActive bool `json:"nftables_active"` SMTPDenyActive bool `json:"smtp_deny_active"` SMTPDenyActiveSource string `json:"smtp_deny_active_source,omitempty"` SMTPDenyBundleAgeSeconds int64 `json:"smtp_deny_bundle_age_seconds"` SMTPDenyBundleRemainingSeconds int64 `json:"smtp_deny_bundle_remaining_seconds"` SMTPDenyCached bool `json:"smtp_deny_cached"` SMTPDenyPolicyValid bool `json:"smtp_deny_policy_valid"` SMTPDenyRefreshCompletedAgeSeconds int64 `json:"smtp_deny_refresh_completed_age_seconds"` SMTPDenyRefreshSuccessful bool `json:"smtp_deny_refresh_successful"` SMTPDenyRemoteCurrentAgeSeconds int64 `json:"smtp_deny_remote_current_age_seconds"` WebAllowActive bool `json:"web_allow_active"` WebAllowCached bool `json:"web_allow_cached"` WireGuardActive bool `json:"wireguard_active"` } type telemetrySnapshot struct { GeneratedAt time.Time `json:"generated_at"` Status relayHealth `json:"status"` } type qdiscDocumentItem struct { Drops uint64 `json:"drops"` Parent string `json:"parent"` } type nftablesSetDocument struct { Nftables []struct { Set *struct { Family string `json:"family"` Name string `json:"name"` Table string `json:"table"` Elem []struct { Element struct { Value string `json:"val"` Expires int64 `json:"expires"` Counter *struct { Packets uint64 `json:"packets"` } `json:"counter"` } `json:"elem"` } `json:"elem"` } `json:"set,omitempty"` } `json:"nftables"` } type nftablesCounterDocument struct { Nftables []struct { Counter *struct { Name string `json:"name"` Packets uint64 `json:"packets"` } `json:"counter"` } `json:"nftables"` } func main() { configPath := flag.String("config", defaultConfigPath, "relay configuration path") nftPath := flag.String("nft", defaultNftPath, "nft path") snapshotPath := flag.String("snapshot", defaultSnapshotPath, "snapshot path") tcPath := flag.String("tc", defaultTCPath, "tc path") telemetryPath := flag.String("telemetry", defaultTelemetryPath, "telemetry snapshot path") once := flag.Bool("once", false, "write one snapshot and exit") flag.Parse() configuration, err := loadConfig(*configPath) if err != nil { fatal(err) } if err := run(context.Background(), configuration, *nftPath, *tcPath, *telemetryPath, *snapshotPath, *once); err != nil { fatal(err) } } func loadConfig(path string) (relayConfig, error) { contents, err := os.ReadFile(path) if err != nil { return relayConfig{}, err } var configuration relayConfig decoder := json.NewDecoder(bytes.NewReader(contents)) decoder.DisallowUnknownFields() if err := decoder.Decode(&configuration); err != nil { return relayConfig{}, err } if err := decoder.Decode(&struct{}{}); !errors.Is(err, io.EOF) { return relayConfig{}, errors.New("SMTP evidence configuration has trailing data") } if configuration.MaximumRecords == 0 || configuration.ObservationWindowSeconds <= 0 { return relayConfig{}, errors.New("SMTP evidence bounds must be positive") } if configuration.ConntrackEmissionBurst <= 0 || configuration.ConntrackEmissionRatePerSecond <= 0 || configuration.ConntrackEventBuffer <= 0 || configuration.ConntrackSampleDenominator == 0 { return relayConfig{}, errors.New("conntrack observation bounds must be positive") } if configuration.ConntrackEventBuffer > 65536 || configuration.ConntrackEmissionBurst > 1000 || configuration.ConntrackEmissionRatePerSecond > 100 { return relayConfig{}, errors.New("conntrack observation bounds exceed tested limits") } if len(configuration.ConntrackServices) == 0 { return relayConfig{}, errors.New("at least one conntrack service is required") } serviceNames := make(map[string]bool) servicePorts := make(map[uint16]bool) for _, service := range configuration.ConntrackServices { if service.Name == "" || serviceNames[service.Name] || !service.TargetAddress.Is4() || service.TargetAddress.IsUnspecified() || len(service.Ports) == 0 { return relayConfig{}, errors.New("conntrack service is invalid") } serviceNames[service.Name] = true for _, port := range service.Ports { if port == 0 || servicePorts[port] { return relayConfig{}, errors.New("conntrack service ports must be nonzero and unique") } servicePorts[port] = true } } interval := configuration.SnapshotIntervalSeconds if interval != 30 && interval != 60 { return relayConfig{}, errors.New("SMTP evidence snapshot interval must be 30 or 60 seconds") } listenAddress, err := netip.ParseAddrPort(configuration.MetricsListenAddress) if err != nil || !listenAddress.Addr().Is4() || listenAddress.Addr().IsUnspecified() || listenAddress.Port() == 0 || listenAddress.String() != configuration.MetricsListenAddress { return relayConfig{}, errors.New("metrics listen address must contain a canonical IPv4 address and port") } return configuration, nil } func run(ctx context.Context, configuration relayConfig, nftPath, tcPath, telemetryPath, snapshotPath string, once bool) error { var latest atomic.Pointer[metricsSnapshot] if err := collect(ctx, configuration, nftPath, tcPath, telemetryPath, snapshotPath, &latest); err != nil { return err } if once { return nil } listener, err := net.Listen("tcp4", configuration.MetricsListenAddress) if err != nil { return err } var conntrackState conntrackMetrics server := &http.Server{Handler: metricsHandler{Conntrack: &conntrackState, Latest: &latest, SMTPConfigured: conntrackServiceConfigured(configuration.ConntrackServices, "smtp")}, ReadHeaderTimeout: 5 * time.Second} serverErrors := make(chan error, 1) observerErrors := make(chan error, 1) go func() { serverErrors <- server.Serve(listener) }() go func() { observerErrors <- startConntrackObserver(ctx, configuration, &conntrackState, os.Stdout) }() defer server.Close() for { timer := time.NewTimer(time.Duration(configuration.SnapshotIntervalSeconds) * time.Second) select { case <-ctx.Done(): timer.Stop() return ctx.Err() case err := <-serverErrors: timer.Stop() return err case err := <-observerErrors: timer.Stop() return err case <-timer.C: if err := collect(ctx, configuration, nftPath, tcPath, telemetryPath, snapshotPath, &latest); err != nil { return err } } } } func collect(ctx context.Context, configuration relayConfig, nftPath, tcPath, telemetryPath, snapshotPath string, latest *atomic.Pointer[metricsSnapshot]) error { now := time.Now().UTC() services := make([]serviceMetrics, 0, len(evidenceSets)) var smtpRecords []evidenceRecord for _, set := range evidenceSets { output, err := exec.CommandContext(ctx, nftPath, "-j", "list", "set", nftablesFamily, nftablesTable, set.Name).Output() if err != nil { return err } records, total, err := parseRecords(now, output, set.Name, configuration.MaximumRecords) if err != nil { return err } services = append(services, serviceMetrics{Records: records, Service: set.Service, Total: total}) if set.Service == "smtp" { smtpRecords = records } } counterOutput, err := exec.CommandContext(ctx, nftPath, "-j", "list", "counters").Output() if err != nil { return err } drops, err := parseDropMetrics(counterOutput, conntrackServiceConfigured(configuration.ConntrackServices, "smtp")) if err != nil { return err } if err := writeSnapshot(snapshotPath, snapshot{ SchemaVersion: 1, GeneratedAt: now, Reason: reason, ObservationWindowSeconds: configuration.ObservationWindowSeconds, Records: smtpRecords, }); err != nil { return err } qdiscOutput, err := exec.CommandContext(ctx, tcPath, "-s", "-j", "qdisc", "show", "dev", "wg-oci0").Output() if err != nil { return err } qdiscDrops, err := parseQdiscMetrics(qdiscOutput) if err != nil { return err } telemetryContents, err := os.ReadFile(telemetryPath) if err != nil { return err } telemetry, err := parseTelemetrySnapshot(telemetryContents) if err != nil { return err } latest.Store(&metricsSnapshot{Drops: drops, GeneratedAt: now, Health: telemetry.Status, QdiscDrops: qdiscDrops, Services: services, TelemetryAt: telemetry.GeneratedAt}) return nil } func parseRecords(now time.Time, contents []byte, expectedSet string, maximum uint64) ([]evidenceRecord, uint64, error) { var document nftablesSetDocument decoder := json.NewDecoder(bytes.NewReader(contents)) if err := decoder.Decode(&document); err != nil { return nil, 0, err } records := make([]evidenceRecord, 0) found := false seen := make(map[netip.Addr]bool) for _, object := range document.Nftables { if object.Set == nil { continue } if object.Set.Family != nftablesFamily || object.Set.Table != nftablesTable || object.Set.Name != expectedSet { return nil, 0, fmt.Errorf("nftables returned the wrong quarantine set for %q", expectedSet) } found = true for _, item := range object.Set.Elem { if item.Element.Counter == nil { return nil, 0, fmt.Errorf("evidence counter is absent for address %q", item.Element.Value) } address, err := netip.ParseAddr(item.Element.Value) if err != nil || !address.Is4() || address.String() != item.Element.Value || seen[address] { return nil, 0, fmt.Errorf("invalid or duplicate evidence address %q", item.Element.Value) } if item.Element.Expires <= 0 { return nil, 0, fmt.Errorf("invalid expiry for evidence address %q", item.Element.Value) } seen[address] = true records = append(records, evidenceRecord{ Address: address.String(), Count: item.Element.Counter.Packets, ExpiresAt: now.Add(time.Duration(item.Element.Expires) * time.Second), }) } } if !found { return nil, 0, fmt.Errorf("quarantine set %q is absent from nftables output", expectedSet) } sort.Slice(records, func(left, right int) bool { leftAddress := netip.MustParseAddr(records[left].Address) rightAddress := netip.MustParseAddr(records[right].Address) return leftAddress.Less(rightAddress) }) total := uint64(len(records)) if total > maximum { records = records[:maximum] } return records, total, nil } func parseDropMetrics(contents []byte, mailConfigured bool) ([]dropMetric, error) { var document nftablesCounterDocument if err := json.Unmarshal(contents, &document); err != nil { return nil, err } values := make(map[string]uint64) for _, object := range document.Nftables { if object.Counter != nil { values[object.Counter.Name] = object.Counter.Packets } } metrics := make([]dropMetric, 0, len(dropMetricSpecs)) for _, spec := range dropMetricSpecs { packets, found := values[spec.Name] if spec.MailOnly && !mailConfigured && !found { packets = 0 found = true } if !found { return nil, fmt.Errorf("nftables counter %q is absent", spec.Name) } metrics = append(metrics, dropMetric{Packets: packets, Reason: spec.Reason, Service: spec.Service}) } return metrics, nil } func parseQdiscMetrics(contents []byte) ([]qdiscMetric, error) { var document []qdiscDocumentItem if err := json.Unmarshal(contents, &document); err != nil { return nil, err } channels := map[string]string{ "1:10": "web", "1:20": "smtp", "1:30": "authenticated_mail", "1:40": "default", } values := map[string]uint64{ "authenticated_mail": 0, "default": 0, "smtp": 0, "total": 0, "web": 0, } for _, qdisc := range document { values["total"] += qdisc.Drops if channel, ok := channels[qdisc.Parent]; ok { values[channel] += qdisc.Drops } } metrics := make([]qdiscMetric, 0, len(values)) for _, channel := range []string{"web", "smtp", "authenticated_mail", "default", "total"} { metrics = append(metrics, qdiscMetric{Channel: channel, Packets: values[channel]}) } return metrics, nil } func parseTelemetrySnapshot(contents []byte) (telemetrySnapshot, error) { var value telemetrySnapshot decoder := json.NewDecoder(bytes.NewReader(contents)) decoder.DisallowUnknownFields() if err := decoder.Decode(&value); err != nil { return telemetrySnapshot{}, err } if err := decoder.Decode(&struct{}{}); !errors.Is(err, io.EOF) { return telemetrySnapshot{}, errors.New("telemetry snapshot has trailing data") } if value.GeneratedAt.IsZero() || value.GeneratedAt.After(time.Now().Add(5*time.Minute)) { return telemetrySnapshot{}, errors.New("telemetry snapshot timestamp is invalid") } return value, nil } type metricsHandler struct { Conntrack *conntrackMetrics Latest *atomic.Pointer[metricsSnapshot] SMTPConfigured bool } func (handler metricsHandler) ServeHTTP(response http.ResponseWriter, request *http.Request) { if request.URL.Path != "/metrics" { http.NotFound(response, request) return } latest := handler.Latest.Load() if latest == nil { http.Error(response, "metrics are not ready", http.StatusServiceUnavailable) return } response.Header().Set("Content-Type", "text/plain; version=0.0.4") fmt.Fprintln(response, "# TYPE oci_l4_relay_tcp_quarantine_packets gauge") for _, service := range latest.Services { for _, record := range service.Records { fmt.Fprintf(response, "oci_l4_relay_tcp_quarantine_packets{service=%q,address=%q} %d\n", service.Service, record.Address, record.Count) } } fmt.Fprintln(response, "# TYPE oci_l4_relay_tcp_quarantine_expires_timestamp_seconds gauge") for _, service := range latest.Services { for _, record := range service.Records { fmt.Fprintf(response, "oci_l4_relay_tcp_quarantine_expires_timestamp_seconds{service=%q,address=%q} %d\n", service.Service, record.Address, record.ExpiresAt.Unix()) } } fmt.Fprintln(response, "# TYPE oci_l4_relay_tcp_evidence_records gauge") for _, service := range latest.Services { fmt.Fprintf(response, "oci_l4_relay_tcp_evidence_records{service=%q} %d\n", service.Service, len(service.Records)) } fmt.Fprintln(response, "# TYPE oci_l4_relay_tcp_evidence_truncated gauge") for _, service := range latest.Services { truncated := 0 if service.Total > uint64(len(service.Records)) { truncated = 1 } fmt.Fprintf(response, "oci_l4_relay_tcp_evidence_truncated{service=%q} %d\n", service.Service, truncated) } fmt.Fprintln(response, "# TYPE oci_l4_relay_tcp_evidence_generated_timestamp_seconds gauge") fmt.Fprintf(response, "oci_l4_relay_tcp_evidence_generated_timestamp_seconds %d\n", latest.GeneratedAt.Unix()) fmt.Fprintln(response, "# TYPE oci_l4_relay_nftables_dropped_packets_total counter") for _, metric := range latest.Drops { fmt.Fprintf(response, "oci_l4_relay_nftables_dropped_packets_total{service=%q,reason=%q} %d\n", metric.Service, metric.Reason, metric.Packets) } fmt.Fprintln(response, "# TYPE oci_l4_relay_qdisc_dropped_packets_total counter") for _, metric := range latest.QdiscDrops { fmt.Fprintf(response, "oci_l4_relay_qdisc_dropped_packets_total{channel=%q} %d\n", metric.Channel, metric.Packets) } fmt.Fprintln(response, "# TYPE oci_l4_relay_component_healthy gauge") for component, healthy := range map[string]bool{ "nftables": latest.Health.NftablesActive, "smtp_deny": !handler.SMTPConfigured || latest.Health.SMTPDenyActive && latest.Health.SMTPDenyPolicyValid, "web_allow": latest.Health.WebAllowActive && latest.Health.WebAllowCached, "wireguard": latest.Health.WireGuardActive, } { fmt.Fprintf(response, "oci_l4_relay_component_healthy{component=%q} %d\n", component, booleanValue(healthy)) } fmt.Fprintln(response, "# TYPE oci_l4_relay_admission_cached gauge") fmt.Fprintf(response, "oci_l4_relay_admission_cached{service=%q} %d\n", "web", booleanValue(latest.Health.WebAllowCached)) fmt.Fprintf(response, "oci_l4_relay_admission_cached{service=%q} %d\n", "smtp", booleanValue(latest.Health.SMTPDenyCached)) fmt.Fprintln(response, "# TYPE oci_l4_relay_smtp_deny_refresh_successful gauge") fmt.Fprintf(response, "oci_l4_relay_smtp_deny_refresh_successful %d\n", booleanValue(latest.Health.SMTPDenyRefreshSuccessful)) fmt.Fprintln(response, "# TYPE oci_l4_relay_smtp_deny_configured gauge") fmt.Fprintf(response, "oci_l4_relay_smtp_deny_configured %d\n", booleanValue(handler.SMTPConfigured)) for name, value := range map[string]int64{ "oci_l4_relay_conntrack_entries": latest.Health.ConntrackCount, "oci_l4_relay_conntrack_maximum": latest.Health.ConntrackMaximum, "oci_l4_relay_smtp_deny_bundle_age_seconds": latest.Health.SMTPDenyBundleAgeSeconds, "oci_l4_relay_smtp_deny_bundle_remaining_seconds": latest.Health.SMTPDenyBundleRemainingSeconds, "oci_l4_relay_smtp_deny_refresh_completed_age_seconds": latest.Health.SMTPDenyRefreshCompletedAgeSeconds, "oci_l4_relay_smtp_deny_remote_current_age_seconds": latest.Health.SMTPDenyRemoteCurrentAgeSeconds, "oci_l4_relay_wireguard_handshake_age_seconds": latest.Health.HandshakeAgeSeconds, "oci_l4_relay_telemetry_generated_timestamp_seconds": latest.TelemetryAt.Unix(), } { fmt.Fprintf(response, "# TYPE %s gauge\n%s %d\n", name, name, value) } if handler.Conntrack != nil { fmt.Fprintln(response, "# TYPE oci_l4_relay_conntrack_observer_events_total counter") for outcome, value := range map[string]uint64{ "eligible": handler.Conntrack.Eligible.Load(), "emitted": handler.Conntrack.Emitted.Load(), "rate_dropped": handler.Conntrack.RateDropped.Load(), "sampled_out": handler.Conntrack.SampledOut.Load(), } { fmt.Fprintf(response, "oci_l4_relay_conntrack_observer_events_total{outcome=%q} %d\n", outcome, value) } fmt.Fprintln(response, "# TYPE oci_l4_relay_conntrack_observer_up gauge") fmt.Fprintf(response, "oci_l4_relay_conntrack_observer_up %d\n", booleanValue(handler.Conntrack.Up.Load())) } } func booleanValue(value bool) int { if value { return 1 } return 0 } func conntrackServiceConfigured(services []conntrackService, name string) bool { for _, service := range services { if service.Name == name { return true } } return false } func writeSnapshot(path string, value snapshot) error { contents, err := json.Marshal(value) if err != nil { return err } directory := filepath.Dir(path) temporary, err := os.CreateTemp(directory, ".smtp-evidence-*.tmp") if err != nil { return err } temporaryPath := temporary.Name() defer os.Remove(temporaryPath) if err := temporary.Chmod(0600); err != nil { temporary.Close() return err } if _, err := temporary.Write(append(contents, '\n')); err != nil { temporary.Close() return err } if err := temporary.Sync(); err != nil { temporary.Close() return err } if err := temporary.Close(); err != nil { return err } return os.Rename(temporaryPath, path) } func fatal(err error) { fmt.Fprintln(os.Stderr, err) os.Exit(1) }