observatory/internal/collector/collector.go

// SPDX-License-Identifier: AGPL-3.0-only

package collector

import (
	"bytes"
	"encoding/json"
	"errors"
	"io"
	"net/netip"
	"net/url"
	"regexp"
	"strconv"
	"strings"
	"time"
	"unicode/utf8"

	"gamertan.com/observatory/internal/model"
)

const MaxLineBytes = 1 << 20

var tendSafeValue = regexp.MustCompile(`^[A-Za-z0-9._:/@+-]{1,256}$`)
var tendOperationID = regexp.MustCompile(`^[0-9a-f]{32}$`)
var tendArtifactDigest = regexp.MustCompile(`^[0-9a-f]{64}$`)
var tendCommit = regexp.MustCompile(`^[0-9a-f]{40}$`)

func Parse(kind string, line []byte, fallback time.Time, sensitiveFields ...string) (model.Signal, model.Observation, error) {
	if len(line) == 0 || len(line) > MaxLineBytes || !utf8.Valid(line) || bytes.IndexByte(line, 0) >= 0 {
		return "", model.Observation{}, errors.New("collector line is outside accepted bounds")
	}
	switch kind {
	case "caddy_json":
		observation, err := parseCaddy(line, fallback, sensitiveFields)
		return model.SignalLogs, observation, err
	case "requestlog_jsonl":
		observation, err := parseRequestLog(line, fallback, sensitiveFields)
		return model.SignalLogs, observation, err
	case "tend_events_jsonl":
		observation, err := parseTend(line, fallback)
		return model.SignalDeployments, observation, err
	default:
		return "", model.Observation{}, errors.New("unsupported collector kind")
	}
}

func parseCaddy(line []byte, fallback time.Time, sensitiveFields []string) (model.Observation, error) {
	var record struct {
		Timestamp float64 `json:"ts"`
		Status    int     `json:"status"`
		Size      int64   `json:"size"`
		Duration  float64 `json:"duration"`
		RequestID string  `json:"request_id"`
		ClientIP  string  `json:"client_ip"`
		Referrer  string  `json:"referrer"`
		UserAgent string  `json:"user_agent"`
		Request   struct {
			Method  string              `json:"method"`
			URI     string              `json:"uri"`
			Headers map[string][]string `json:"headers"`
		} `json:"request"`
	}
	if err := json.Unmarshal(line, &record); err != nil {
		return model.Observation{}, errors.New("invalid Caddy JSON record")
	}
	timestamp := fallback
	if record.Timestamp > 0 {
		seconds, fraction := mathModf(record.Timestamp)
		timestamp = time.Unix(seconds, int64(fraction*1e9)).UTC()
	}
	path := "/"
	parsed, parseErr := url.ParseRequestURI(record.Request.URI)
	if parseErr == nil && parsed.Path != "" {
		path = parsed.EscapedPath()
	}
	attributes := map[string]string{
		"http.method":         bounded(record.Request.Method, 32),
		"http.path":           bounded(path, 4096),
		"http.status_code":    strconv.Itoa(record.Status),
		"http.response_bytes": strconv.FormatInt(record.Size, 10),
		"duration_ns":         strconv.FormatInt(int64(record.Duration*1e9), 10),
	}
	requestID := record.RequestID
	if !safeIdentifier(requestID) {
		requestID = firstHeader(record.Request.Headers, "X-Request-Id", "X-Request-ID")
	}
	if safeIdentifier(requestID) {
		attributes["request.id"] = requestID
	}
	selected := sensitiveFieldSet(sensitiveFields)
	if selected["client_ip"] {
		if address, err := netip.ParseAddr(record.ClientIP); err == nil && address.IsValid() && address.Zone() == "" {
			attributes["client.address"] = address.String()
		}
	}
	if selected["query"] && parseErr == nil && parsed.RawQuery != "" {
		attributes["url.query"] = bounded(parsed.RawQuery, 4096)
	}
	if selected["referrer"] {
		if value := boundedSensitive(record.Referrer, 4096); value != "" {
			attributes["http.request.referrer"] = value
		}
	}
	if selected["user_agent"] {
		if value := boundedSensitive(record.UserAgent, 1024); value != "" {
			attributes["user_agent.original"] = value
		}
	}
	return model.Observation{Timestamp: timestamp, Name: "caddy.http.request", CorrelationID: attributes["request.id"], Attributes: attributes}, nil
}

func parseRequestLog(line []byte, fallback time.Time, sensitiveFields []string) (model.Observation, error) {
	var raw map[string]json.RawMessage
	if err := json.Unmarshal(line, &raw); err != nil {
		return model.Observation{}, errors.New("invalid requestlog JSON record")
	}
	timestamp := parseTimeFields(raw, fallback, "observed_at", "timestamp", "time")
	attributes := map[string]string{}
	copyString(raw, attributes, "method", "http.method", 32)
	copyString(raw, attributes, "route", "http.route", 4096)
	copyNumber(raw, attributes, "status", "http.status_code")
	copyNumber(raw, attributes, "bytes", "http.response_bytes")
	copyNumber(raw, attributes, "duration_ns", "duration_ns")
	copyString(raw, attributes, "authorization_outcome", "auth.outcome", 128)
	requestID := rawString(raw["request_id"], 128)
	if safeIdentifier(requestID) {
		attributes["request.id"] = requestID
	}
	selected := sensitiveFieldSet(sensitiveFields)
	if selected["client_ip"] {
		if address, err := netip.ParseAddr(rawString(raw["client_ip"], 64)); err == nil && address.IsValid() && address.Zone() == "" {
			attributes["client.address"] = address.String()
		}
	}
	if selected["query"] {
		copyString(raw, attributes, "query", "url.query", 4096)
	}
	if selected["referrer"] {
		copyString(raw, attributes, "referer", "http.request.referrer", 2048)
	}
	if selected["user_agent"] {
		copyString(raw, attributes, "user_agent", "user_agent.original", 1024)
	}
	if selected["session_id"] {
		copyString(raw, attributes, "session_id", "session.id", 256)
	}
	return model.Observation{Timestamp: timestamp, Name: "application.http.request", CorrelationID: requestID, Attributes: attributes}, nil
}

func sensitiveFieldSet(fields []string) map[string]bool {
	selected := make(map[string]bool, len(fields))
	for _, field := range fields {
		selected[field] = true
	}
	return selected
}

func boundedSensitive(value string, maximum int) string {
	if value == "" || len(value) > maximum || !utf8.ValidString(value) || strings.ContainsAny(value, "\x00\r\n") {
		return ""
	}
	return value
}

func parseTend(line []byte, _ time.Time) (model.Observation, error) {
	if len(line) > 4096 {
		return model.Observation{}, errors.New("invalid Tend deployment event")
	}
	var record tendEvent
	decoder := json.NewDecoder(bytes.NewReader(line))
	decoder.DisallowUnknownFields()
	if err := decoder.Decode(&record); err != nil || record.validate() != nil {
		return model.Observation{}, errors.New("invalid Tend deployment event")
	}
	if err := decoder.Decode(&struct{}{}); !errors.Is(err, io.EOF) {
		return model.Observation{}, errors.New("invalid Tend deployment event")
	}
	timestamp, _ := time.Parse(time.RFC3339Nano, record.ObservedAt)
	attributes := map[string]string{
		"deployment.operation_id": record.OperationID,
		"service.name":            record.Service,
		"deployment.artifact":     record.ArtifactDigest,
		"deployment.commit":       record.Commit,
		"deployment.version":      record.ReleaseVersion,
		"deployment.phase":        record.Phase,
		"deployment.duration_ms":  strconv.FormatInt(record.DurationMillis, 10),
		"deployment.outcome":      record.Outcome,
	}
	if record.Slot != "" {
		attributes["deployment.slot"] = record.Slot
	}
	return model.Observation{Timestamp: timestamp.UTC(), Name: "tend.deployment", CorrelationID: record.OperationID, Attributes: attributes}, nil
}

type tendEvent struct {
	Version        int    `json:"version"`
	OperationID    string `json:"operation_id"`
	Service        string `json:"service"`
	ArtifactDigest string `json:"artifact_digest"`
	Commit         string `json:"commit"`
	ReleaseVersion string `json:"release_version"`
	Phase          string `json:"phase"`
	Slot           string `json:"slot"`
	DurationMillis int64  `json:"duration_ms"`
	Outcome        string `json:"outcome"`
	ObservedAt     string `json:"observed_at"`
}

func (event tendEvent) validate() error {
	if event.Version != 1 || !tendOperationID.MatchString(event.OperationID) || !tendArtifactDigest.MatchString(event.ArtifactDigest) || !tendCommit.MatchString(event.Commit) {
		return errors.New("identity or provenance is invalid")
	}
	for _, value := range []string{event.Service, event.ArtifactDigest, event.Commit, event.ReleaseVersion, event.Phase, event.Outcome} {
		if !tendSafeValue.MatchString(value) || strings.ContainsRune(value, '\x00') {
			return errors.New("value is invalid")
		}
	}
	if event.Slot != "" && !tendSafeValue.MatchString(event.Slot) {
		return errors.New("slot is invalid")
	}
	if event.DurationMillis < 0 {
		return errors.New("duration is invalid")
	}
	if _, err := time.Parse(time.RFC3339Nano, event.ObservedAt); err != nil {
		return errors.New("timestamp is invalid")
	}
	return nil
}

func copyString(raw map[string]json.RawMessage, attributes map[string]string, source, target string, maximum int) {
	if value := rawString(raw[source], maximum); value != "" {
		attributes[target] = value
	}
}

func copyNumber(raw map[string]json.RawMessage, attributes map[string]string, source, target string) {
	value := strings.TrimSpace(string(raw[source]))
	if value == "" || len(value) > 64 {
		return
	}
	if _, err := strconv.ParseFloat(value, 64); err == nil {
		attributes[target] = value
	}
}

func rawString(raw json.RawMessage, maximum int) string {
	var value string
	if len(raw) == 0 || json.Unmarshal(raw, &value) != nil || len(value) > maximum || !utf8.ValidString(value) || strings.ContainsAny(value, "\x00\r\n") {
		return ""
	}
	return value
}

func parseTimeFields(raw map[string]json.RawMessage, fallback time.Time, names ...string) time.Time {
	for _, name := range names {
		value := rawString(raw[name], 128)
		if value == "" {
			continue
		}
		if parsed, err := time.Parse(time.RFC3339Nano, value); err == nil {
			return parsed.UTC()
		}
	}
	return fallback.UTC()
}

func firstHeader(headers map[string][]string, names ...string) string {
	for _, name := range names {
		for key, values := range headers {
			if strings.EqualFold(key, name) && len(values) == 1 {
				return values[0]
			}
		}
	}
	return ""
}

func safeIdentifier(value string) bool {
	if len(value) < 1 || len(value) > 128 {
		return false
	}
	for _, r := range value {
		if !(r >= 'a' && r <= 'z' || r >= 'A' && r <= 'Z' || r >= '0' && r <= '9' || strings.ContainsRune("._:-", r)) {
			return false
		}
	}
	return true
}

func bounded(value string, maximum int) string {
	if len(value) > maximum {
		for maximum > 0 && !utf8.ValidString(value[:maximum]) {
			maximum--
		}
		return value[:maximum]
	}
	return value
}

func mathModf(value float64) (int64, float64) {
	seconds := int64(value)
	return seconds, value - float64(seconds)
}