diff --git a/internal/events/event.go b/internal/events/event.go new file mode 100644 index 0000000..190f1c8 --- /dev/null +++ b/internal/events/event.go @@ -0,0 +1,58 @@ +// Package events is the `harness events` webhook receiver (DESIGN "Events +// are V1 scope"): Redmine, Discourse and Gitea webhooks land here, are +// verified against per-source secrets, normalized into one internal Event +// record, persisted append-only (JSONL, dedup by provider event id), and +// mapped to a conductor action. Secret material is never logged and never +// persisted; only payload digests are. +package events + +import "time" + +// Provider sources. +const ( + SourceRedmine = "redmine" + SourceDiscourse = "discourse" + SourceGitea = "gitea" +) + +// Actions an event can map to (DESIGN trigger -> action semantics). +const ( + ActionDispatchTurn = "dispatch_turn" // redmine issue update/note -> run a turn for that stack + ActionRespondTurn = "respond_turn" // discourse post reply -> context update + response turn + ActionPipelineStep = "pipeline_step" // gitea PR approved/merged -> next pipeline step + ActionIgnore = "ignore" // recognized but not actionable +) + +// Event is the normalized internal record every provider webhook becomes. +// One shape, three sources: source + kind say what happened, actor who did +// it, SubjectID/Subject what it happened to (issue / topic / PR). +type Event struct { + Source string `json:"source"` + Kind string `json:"kind"` + // Action is the mapped conductor action (dispatch_turn, ...). + Action string `json:"action"` + // Actor is the provider-side user who triggered the event. + Actor string `json:"actor"` + // SubjectID canonically identifies the subject: redmine:issue:42, + // discourse:topic:7, gitea:pr:ukrrs/MOPAC#5. + SubjectID string `json:"subject_id"` + // Subject is the human-readable title (best effort). + Subject string `json:"subject,omitempty"` + // Repo is the gitea repository full_name when applicable. + Repo string `json:"repo,omitempty"` + // ProviderID is the provider's own event id (delivery/event header, + // falling back to the payload digest) and the dedup identity. + ProviderID string `json:"provider_id"` + PayloadDigest string `json:"payload_digest"` + ReceivedAt time.Time `json:"received_at"` +} + +// DedupKey is the append-store identity: source + provider event id. +func (e Event) DedupKey() string { + return e.Source + ":" + e.ProviderID +} + +// Sources lists the receiver's provider sources. +func Sources() []string { + return []string{SourceRedmine, SourceDiscourse, SourceGitea} +} diff --git a/internal/events/normalize.go b/internal/events/normalize.go new file mode 100644 index 0000000..8af36bd --- /dev/null +++ b/internal/events/normalize.go @@ -0,0 +1,276 @@ +package events + +import ( + "bytes" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "fmt" + "net/http" + "strings" + "time" +) + +// Provider delivery/event id headers used for dedup before falling back to +// the payload digest. +const ( + giteaDeliveryHeader = "X-Gitea-Delivery" + discourseEventIDHeader = "X-Discourse-Event-Id" + redmineDeliveryFallback = "X-Redmine-Delivery" // some plugins send it +) + +// Normalize maps a verified provider payload to the internal Event record. +// It is deliberately tolerant: providers (and their plugin ecosystems) +// disagree on payload shapes, so extraction tries the known variants and +// leaves anything missing empty rather than rejecting. Unparseable bodies +// are the only hard error. +func Normalize(source string, header http.Header, body []byte, now time.Time) (Event, error) { + var root map[string]any + dec := json.NewDecoder(bytes.NewReader(body)) + dec.UseNumber() + if err := dec.Decode(&root); err != nil { + return Event{}, fmt.Errorf("payload is not a JSON object: %w", err) + } + + sum := sha256.Sum256(body) + ev := Event{ + Source: source, + PayloadDigest: hex.EncodeToString(sum[:]), + ReceivedAt: now.UTC(), + } + + switch source { + case SourceRedmine: + normalizeRedmine(&ev, header, root) + case SourceDiscourse: + normalizeDiscourse(&ev, header, root) + case SourceGitea: + normalizeGitea(&ev, header, root) + default: + return Event{}, fmt.Errorf("unknown event source %q", source) + } + + // Dedup identity: the provider's event id, else the payload digest + // (identical replays dedup; different payloads never collide). + if ev.ProviderID == "" { + ev.ProviderID = ev.PayloadDigest + } + if ev.Actor == "" { + ev.Actor = "unknown" + } + if ev.Kind == "" { + ev.Kind = "unknown" + } + ev.Action = MapAction(ev.Source, ev.Kind) + return ev, nil +} + +// MapAction implements the DESIGN trigger -> action mapping: +// redmine issue update/note -> dispatch_turn; discourse post reply -> +// respond_turn; gitea PR approved/merged -> pipeline_step; else ignore. +func MapAction(source, kind string) string { + switch source { + case SourceRedmine: + if kind == "issue_updated" || kind == "issue_edited" || + strings.Contains(kind, "note") || strings.Contains(kind, "journal") { + return ActionDispatchTurn + } + case SourceDiscourse: + if kind == "post_created" || kind == "post_edited" || kind == "post" { + return ActionRespondTurn + } + case SourceGitea: + if kind == "pr_approved" || kind == "pr_merged" { + return ActionPipelineStep + } + } + return ActionIgnore +} + +// normalizeRedmine handles the redmine_webhooks shape +// {"event_name": "...", "payload": {"issue": {...}}} and flatter plugin +// payloads that put "issue" (and the actor) at the top level. +func normalizeRedmine(ev *Event, header http.Header, root map[string]any) { + p := obj(root, "payload") + issue := obj(p, "issue") + if issue == nil { + issue = obj(root, "issue") + } + ev.Kind = firstNonEmpty( + str(root, "event_name"), + str(p, "event_name"), + header.Get("X-Redmine-Event"), + ) + ev.Actor = firstNonEmpty( + person(p, "user", "author"), + person(root, "user", "author"), + person(issue, "user", "author"), + ) + if issue != nil { + if ev.Kind == "" { + ev.Kind = "issue_updated" + } + if id := numStr(issue, "id"); id != "" { + ev.SubjectID = "redmine:issue:" + id + } + ev.Subject = str(issue, "subject") + } + ev.ProviderID = firstNonEmpty(header.Get(redmineDeliveryFallback)) +} + +// normalizeDiscourse handles Discourse webhook bodies +// {"post": {"id":.., "topic_id":.., "username":.., ...}}; the event type +// rides in the X-Discourse-Event header (e.g. post_created). +func normalizeDiscourse(ev *Event, header http.Header, root map[string]any) { + ev.Kind = firstNonEmpty( + header.Get("X-Discourse-Event"), + header.Get("X-Discourse-Event-Type"), + str(root, "event_type"), + ) + post := obj(root, "post") + if post != nil { + if ev.Kind == "" { + ev.Kind = "post_created" + } + if id := numStr(post, "topic_id", "topicId"); id != "" { + ev.SubjectID = "discourse:topic:" + id + } else if id := numStr(root, "topic_id"); id != "" { + ev.SubjectID = "discourse:topic:" + id + } + ev.Subject = str(post, "topic_title", "topic_slug") + ev.Actor = firstNonEmpty(str(post, "username"), person(root, "user"), str(root, "username")) + } else { + if id := numStr(root, "topic_id"); id != "" { + ev.SubjectID = "discourse:topic:" + id + } + ev.Actor = firstNonEmpty(person(root, "user"), str(root, "username")) + } + ev.ProviderID = firstNonEmpty(header.Get(discourseEventIDHeader), str(root, "id")) +} + +// normalizeGitea handles Gitea webhook bodies. PR events carry +// {"action": ..., "number": N, "pull_request": {...}, "repository": {...}, +// "sender": {...}}; issue events mirror that with "issue". kind vocabulary: +// pr_approved / pr_merged / pr_opened / ... and issue_*. +func normalizeGitea(ev *Event, header http.Header, root map[string]any) { + ev.ProviderID = header.Get(giteaDeliveryHeader) + action := str(root, "action") + if repo := obj(root, "repository"); repo != nil { + ev.Repo = str(repo, "full_name") + } + ev.Actor = person(root, "sender") + + pr := obj(root, "pull_request") + issue := obj(root, "issue") + switch { + case pr != nil: + kind := action + if action == "closed" && truthy(pr, "merged") { + kind = "merged" + } + if kind == "" { + kind = "unknown" + } + ev.Kind = "pr_" + kind + if n := firstNonEmpty(numStr(root, "number"), numStr(pr, "number")); n != "" { + ev.SubjectID = giteaRef(ev.Repo, "pr", n) + } + ev.Subject = str(pr, "title") + case issue != nil: + if action == "" { + action = "updated" + } + ev.Kind = "issue_" + action + if n := firstNonEmpty(numStr(root, "number"), numStr(issue, "number"), numStr(issue, "id")); n != "" { + ev.SubjectID = giteaRef(ev.Repo, "issue", n) + } + ev.Subject = str(issue, "title") + default: + ev.Kind = firstNonEmpty( + header.Get("X-Gitea-Event-Type"), + header.Get("X-Gitea-Event"), + action, + "unknown", + ) + } +} + +func giteaRef(repo, kind, n string) string { + if repo == "" { + return "gitea:" + kind + ":" + n + } + return "gitea:" + kind + ":" + repo + "#" + n +} + +// --- tolerant JSON extraction helpers ------------------------------------- + +func obj(m map[string]any, key string) map[string]any { + if m == nil { + return nil + } + if v, ok := m[key].(map[string]any); ok { + return v + } + return nil +} + +// str returns the first non-empty string among the keys. +func str(m map[string]any, keys ...string) string { + if m == nil { + return "" + } + for _, k := range keys { + if v, ok := m[k].(string); ok && v != "" { + return v + } + } + return "" +} + +// numStr returns the first present numeric (or string) key rendered as a +// decimal string; ids must survive without float mangling. +func numStr(m map[string]any, keys ...string) string { + if m == nil { + return "" + } + for _, k := range keys { + switch v := m[k].(type) { + case json.Number: + return v.String() + case string: + if v != "" { + return v + } + case float64: + return fmt.Sprintf("%d", int64(v)) + } + } + return "" +} + +// person extracts an actor from nested user-ish objects: login, name, +// username, full_name in order. +func person(m map[string]any, keys ...string) string { + for _, k := range keys { + if o := obj(m, k); o != nil { + if s := str(o, "login", "username", "name", "full_name"); s != "" { + return s + } + } + } + return "" +} + +func truthy(m map[string]any, key string) bool { + v, ok := m[key].(bool) + return ok && v +} + +func firstNonEmpty(vals ...string) string { + for _, v := range vals { + if v != "" { + return v + } + } + return "" +} diff --git a/internal/events/normalize_test.go b/internal/events/normalize_test.go new file mode 100644 index 0000000..66573bb --- /dev/null +++ b/internal/events/normalize_test.go @@ -0,0 +1,236 @@ +package events + +import ( + "net/http" + "strings" + "testing" + "time" +) + +var testNow = time.Date(2026, 8, 28, 22, 0, 0, 0, time.UTC) + +func TestNormalize(t *testing.T) { + cases := []struct { + name string + source string + headers map[string]string + body string + want Event + wantErr string + }{ + { + name: "gitea pr approved", + source: SourceGitea, + headers: map[string]string{ + giteaDeliveryHeader: "d-1234", + }, + body: `{"action":"approved","number":5, + "pull_request":{"number":5,"title":"Add events receiver","merged":false}, + "repository":{"full_name":"ukrrs/MOPAC"}, + "sender":{"login":"charles"}}`, + want: Event{ + Source: SourceGitea, Kind: "pr_approved", Action: ActionPipelineStep, + Actor: "charles", SubjectID: "gitea:pr:ukrrs/MOPAC#5", + Subject: "Add events receiver", Repo: "ukrrs/MOPAC", + ProviderID: "d-1234", + }, + }, + { + name: "gitea pr closed and merged maps to pr_merged", + source: SourceGitea, + body: `{"action":"closed","number":7, + "pull_request":{"number":7,"title":"Fix loop","merged":true}, + "repository":{"full_name":"ukrrs/MOPAC"},"sender":{"login":"alice"}}`, + want: Event{ + Source: SourceGitea, Kind: "pr_merged", Action: ActionPipelineStep, + Actor: "alice", SubjectID: "gitea:pr:ukrrs/MOPAC#7", Subject: "Fix loop", Repo: "ukrrs/MOPAC", + }, + }, + { + name: "gitea pr opened is stored but ignored", + source: SourceGitea, + body: `{"action":"opened","number":9, + "pull_request":{"number":9,"title":"New"},"sender":{"login":"bob"}}`, + want: Event{ + Source: SourceGitea, Kind: "pr_opened", Action: ActionIgnore, + Actor: "bob", SubjectID: "gitea:pr:9", Subject: "New", + }, + }, + { + name: "gitea push event", + source: SourceGitea, + headers: map[string]string{ + "X-Gitea-Event": "push", + giteaDeliveryHeader: "d-push", + }, + body: `{"ref":"refs/heads/main","sender":{"login":"ci"}}`, + want: Event{ + Source: SourceGitea, Kind: "push", Action: ActionIgnore, + Actor: "ci", ProviderID: "d-push", + }, + }, + { + name: "redmine webhook plugin shape", + source: SourceRedmine, + body: `{"event_name":"issue_updated", + "payload":{"issue":{"id":42,"subject":"Ship phase 2b","author":{"name":"charles"}}, + "user":{"login":"charles"}}}`, + want: Event{ + Source: SourceRedmine, Kind: "issue_updated", Action: ActionDispatchTurn, + Actor: "charles", SubjectID: "redmine:issue:42", Subject: "Ship phase 2b", + }, + }, + { + name: "redmine flat shape defaults kind", + source: SourceRedmine, + body: `{"issue":{"id":43,"subject":"Flat plugin","author":{"login":"dana"}}}`, + want: Event{ + Source: SourceRedmine, Kind: "issue_updated", Action: ActionDispatchTurn, + Actor: "dana", SubjectID: "redmine:issue:43", Subject: "Flat plugin", + }, + }, + { + name: "redmine journal event dispatches", + source: SourceRedmine, + body: `{"event_name":"journal_created","payload":{"issue":{"id":8,"subject":"Note"}}}`, + want: Event{ + Source: SourceRedmine, Kind: "journal_created", Action: ActionDispatchTurn, + SubjectID: "redmine:issue:8", Subject: "Note", Actor: "unknown", + }, + }, + { + name: "discourse post created", + source: SourceDiscourse, + headers: map[string]string{ + "X-Discourse-Event": "post_created", + discourseEventIDHeader: "42", + }, + body: `{"post":{"id":99,"topic_id":7,"username":"charles","topic_title":"Phase 2b plan"}}`, + want: Event{ + Source: SourceDiscourse, Kind: "post_created", Action: ActionRespondTurn, + Actor: "charles", SubjectID: "discourse:topic:7", + Subject: "Phase 2b plan", ProviderID: "42", + }, + }, + { + name: "discourse without headers reads payload", + source: SourceDiscourse, + body: `{"event_type":"post_edited","post":{"topic_id":3,"username":"eve"},"id":"77"}`, + want: Event{ + Source: SourceDiscourse, Kind: "post_edited", Action: ActionRespondTurn, + Actor: "eve", SubjectID: "discourse:topic:3", ProviderID: "77", + }, + }, + { + name: "discourse user event ignored", + source: SourceDiscourse, + headers: map[string]string{"X-Discourse-Event": "user_created"}, + body: `{"user":{"username":"newbie"}}`, + want: Event{ + Source: SourceDiscourse, Kind: "user_created", Action: ActionIgnore, + Actor: "newbie", + }, + }, + { + name: "not json", + source: SourceGitea, + body: `not json`, + wantErr: "payload is not a JSON object", + }, + { + name: "json array not object", + source: SourceRedmine, + body: `[1,2,3]`, + wantErr: "payload is not a JSON object", + }, + { + name: "unknown source", + source: "slack", + body: `{}`, + wantErr: "unknown event source", + }, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + header := http.Header{} + for k, v := range tc.headers { + header.Set(k, v) + } + ev, err := Normalize(tc.source, header, []byte(tc.body), testNow) + if tc.wantErr != "" { + if err == nil || !strings.Contains(err.Error(), tc.wantErr) { + t.Fatalf("error = %v, want containing %q", err, tc.wantErr) + } + return + } + if err != nil { + t.Fatalf("Normalize: %v", err) + } + got := Event{ + Source: ev.Source, Kind: ev.Kind, Action: ev.Action, Actor: ev.Actor, + SubjectID: ev.SubjectID, Subject: ev.Subject, Repo: ev.Repo, + ProviderID: ev.ProviderID, + } + if tc.want.ProviderID == "" { + got.ProviderID = "" // digest fallback is asserted separately + } + if got != tc.want { + t.Errorf("normalized event:\n got %+v\n want %+v", got, tc.want) + } + if ev.ReceivedAt != testNow { + t.Errorf("ReceivedAt = %v, want %v", ev.ReceivedAt, testNow) + } + if len(ev.PayloadDigest) != 64 { + t.Errorf("PayloadDigest = %q, want sha256 hex", ev.PayloadDigest) + } + // No provider id anywhere -> digest fallback (dedup still works). + if tc.want.ProviderID == "" && ev.ProviderID != ev.PayloadDigest { + t.Errorf("ProviderID = %q, want digest fallback", ev.ProviderID) + } + }) + } +} + +func TestMapAction(t *testing.T) { + cases := []struct { + source, kind, want string + }{ + {SourceRedmine, "issue_updated", ActionDispatchTurn}, + {SourceRedmine, "issue_note_added", ActionDispatchTurn}, + {SourceRedmine, "issue_created", ActionIgnore}, + {SourceRedmine, "member_added", ActionIgnore}, + {SourceDiscourse, "post_created", ActionRespondTurn}, + {SourceDiscourse, "post", ActionRespondTurn}, + {SourceDiscourse, "topic_created", ActionIgnore}, + {SourceGitea, "pr_approved", ActionPipelineStep}, + {SourceGitea, "pr_merged", ActionPipelineStep}, + {SourceGitea, "pr_synchronized", ActionIgnore}, + {SourceGitea, "issue_opened", ActionIgnore}, + {SourceGitea, "push", ActionIgnore}, + } + for _, tc := range cases { + if got := MapAction(tc.source, tc.kind); got != tc.want { + t.Errorf("MapAction(%s, %s) = %s, want %s", tc.source, tc.kind, got, tc.want) + } + } +} + +func TestNormalizeDigestStable(t *testing.T) { + body := []byte(`{"action":"approved","number":1,"pull_request":{"number":1}}`) + h := http.Header{giteaDeliveryHeader: []string{"same"}} + a, err := Normalize(SourceGitea, h, body, testNow) + if err != nil { + t.Fatal(err) + } + b, err := Normalize(SourceGitea, h, body, testNow.Add(time.Hour)) + if err != nil { + t.Fatal(err) + } + if a.DedupKey() != b.DedupKey() { + t.Errorf("same delivery must keep one dedup key: %s vs %s", a.DedupKey(), b.DedupKey()) + } + if a.PayloadDigest != b.PayloadDigest { + t.Errorf("digest changed for identical payload") + } +} diff --git a/internal/events/store.go b/internal/events/store.go new file mode 100644 index 0000000..2326bf4 --- /dev/null +++ b/internal/events/store.go @@ -0,0 +1,101 @@ +package events + +import ( + "bufio" + "encoding/json" + "fmt" + "os" + "path/filepath" + "sync" +) + +// Store is the append-only event log: one Event per JSON line under +// stateDir/events.jsonl, deduped by provider event id (source:providerID). +// The dedup index is rebuilt from the file at open, so restarts keep the +// at-least-once semantics (a torn tail line is skipped, not fatal). +type Store struct { + mu sync.Mutex + path string + f *os.File + seen map[string]bool +} + +// OpenStore creates/opens the state dir and the event log. +func OpenStore(stateDir string) (*Store, error) { + if err := os.MkdirAll(stateDir, 0o700); err != nil { + return nil, fmt.Errorf("events state dir: %w", err) + } + path := filepath.Join(stateDir, "events.jsonl") + f, err := os.OpenFile(path, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0o600) + if err != nil { + return nil, fmt.Errorf("open event log: %w", err) + } + s := &Store{path: path, f: f, seen: make(map[string]bool)} + if err := s.loadIndex(); err != nil { + f.Close() + return nil, err + } + return s, nil +} + +// loadIndex populates the dedup set from the existing log. +func (s *Store) loadIndex() error { + rf, err := os.Open(s.path) + if err != nil { + if os.IsNotExist(err) { + return nil + } + return fmt.Errorf("read event log: %w", err) + } + defer rf.Close() + sc := bufio.NewScanner(rf) + sc.Buffer(make([]byte, 0, 64*1024), 1<<20) + for sc.Scan() { + var ev Event + if err := json.Unmarshal(sc.Bytes(), &ev); err != nil { + continue // torn/malformed tail line; dedup stays conservative + } + if ev.ProviderID != "" { + s.seen[ev.DedupKey()] = true + } + } + return sc.Err() +} + +// Append records ev unless its provider event id is already logged. +// stored=false means duplicate (replay); err means the write failed and the +// event is NOT deduped (safe to retry). +func (s *Store) Append(ev Event) (stored bool, err error) { + s.mu.Lock() + defer s.mu.Unlock() + key := ev.DedupKey() + if s.seen[key] { + return false, nil + } + line, err := json.Marshal(ev) + if err != nil { + return false, fmt.Errorf("encode event: %w", err) + } + if _, err := s.f.Write(append(line, '\n')); err != nil { + return false, fmt.Errorf("append event log: %w", err) + } + s.seen[key] = true + return true, nil +} + +// Len reports how many distinct events the index holds (ops/tests). +func (s *Store) Len() int { + s.mu.Lock() + defer s.mu.Unlock() + return len(s.seen) +} + +// Path exposes the log location (ops messages only; contents stay put). +func (s *Store) Path() string { return s.path } + +// Close closes the underlying file. +func (s *Store) Close() error { + s.mu.Lock() + defer s.mu.Unlock() + return s.f.Close() +} diff --git a/internal/events/store_test.go b/internal/events/store_test.go new file mode 100644 index 0000000..81bfea3 --- /dev/null +++ b/internal/events/store_test.go @@ -0,0 +1,114 @@ +package events + +import ( + "os" + "path/filepath" + "strings" + "testing" + "time" +) + +func TestStoreAppendAndDedup(t *testing.T) { + dir := t.TempDir() + s, err := OpenStore(dir) + if err != nil { + t.Fatalf("OpenStore: %v", err) + } + base := Event{ + Source: SourceGitea, Kind: "pr_approved", Action: ActionPipelineStep, + ProviderID: "d-1", PayloadDigest: strings.Repeat("a", 64), ReceivedAt: testNow, + } + + if stored, err := s.Append(base); err != nil || !stored { + t.Fatalf("first append = stored=%v err=%v", stored, err) + } + if stored, err := s.Append(base); err != nil || stored { + t.Fatalf("replay append = stored=%v err=%v (want dup)", stored, err) + } + sameIDOtherSource := base + sameIDOtherSource.Source = SourceRedmine + if stored, err := s.Append(sameIDOtherSource); err != nil || !stored { + t.Fatalf("same provider id, other source = stored=%v err=%v (want stored)", stored, err) + } + if s.Len() != 2 { + t.Errorf("Len = %d, want 2", s.Len()) + } + s.Close() + + data, err := os.ReadFile(filepath.Join(dir, "events.jsonl")) + if err != nil { + t.Fatalf("read log: %v", err) + } + lines := strings.Split(strings.TrimSpace(string(data)), "\n") + if len(lines) != 2 { + t.Fatalf("log has %d lines, want 2:\n%s", len(lines), data) + } + if !strings.Contains(lines[0], `"provider_id":"d-1"`) { + t.Errorf("line 1 missing provider id: %s", lines[0]) + } + if !strings.Contains(lines[0], `"source":"gitea"`) { + t.Errorf("line 1 wrong source: %s", lines[0]) + } +} + +func TestStoreDedupSurvivesRestart(t *testing.T) { + dir := t.TempDir() + s, err := OpenStore(dir) + if err != nil { + t.Fatal(err) + } + ev := Event{Source: SourceDiscourse, Kind: "post_created", ProviderID: "ev-9", ReceivedAt: time.Now()} + if stored, _ := s.Append(ev); !stored { + t.Fatal("first append must store") + } + s.Close() + + s2, err := OpenStore(dir) + if err != nil { + t.Fatal(err) + } + defer s2.Close() + if stored, _ := s2.Append(ev); stored { + t.Fatal("duplicate after restart must not store again") + } + if s2.Len() != 1 { + t.Errorf("Len after restart = %d, want 1", s2.Len()) + } +} + +func TestStoreSkipsTornTailLine(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "events.jsonl") + good := `{"source":"gitea","provider_id":"d-1","kind":"push"}` + "\n" + torn := `{"source":"gitea","provider_id":"d-2","ki` + if err := os.WriteFile(path, []byte(good+torn), 0o600); err != nil { + t.Fatal(err) + } + s, err := OpenStore(dir) + if err != nil { + t.Fatalf("OpenStore with torn tail: %v", err) + } + defer s.Close() + if s.Len() != 1 { + t.Errorf("Len = %d, want 1 (torn line skipped)", s.Len()) + } + if stored, _ := s.Append(Event{Source: SourceGitea, ProviderID: "d-2", ReceivedAt: time.Now()}); !stored { + t.Errorf("event from torn line must be appendable again") + } +} + +func TestStorePermissions(t *testing.T) { + dir := t.TempDir() + s, err := OpenStore(dir) + if err != nil { + t.Fatal(err) + } + s.Close() + fi, err := os.Stat(filepath.Join(dir, "events.jsonl")) + if err != nil { + t.Fatal(err) + } + if fi.Mode().Perm() != 0o600 { + t.Errorf("events.jsonl mode = %v, want 0600", fi.Mode().Perm()) + } +} diff --git a/internal/events/verify.go b/internal/events/verify.go new file mode 100644 index 0000000..cef8b7a --- /dev/null +++ b/internal/events/verify.go @@ -0,0 +1,65 @@ +package events + +import ( + "crypto/hmac" + "crypto/sha256" + "crypto/subtle" + "encoding/hex" + "errors" + "fmt" +) + +// giteaSignatureHeader carries the hex HMAC-SHA256 of the raw body. +const giteaSignatureHeader = "X-Gitea-Signature" + +// Default shared-secret header names (overridable per source in config; +// gitea is always HMAC and ignores the override). +const ( + DefaultRedmineSecretHeader = "X-Redmine-Webhook-Secret" + DefaultDiscourseSecretHeader = "X-Discourse-Webhook-Secret" +) + +// ErrUnverified marks every verification failure so callers reject with 401 +// without distinguishing WHY (no oracle for attackers poking the receiver). +var ErrUnverified = errors.New("webhook not verified") + +// VerifyGiteaHMAC checks the X-Gitea-Signature header (hex HMAC-SHA256 of +// the raw request body) against the shared secret. Comparison is +// constant-time; error text never echoes header values. +func VerifyGiteaHMAC(body []byte, headerValue, secret string) error { + if headerValue == "" { + return fmt.Errorf("%w: missing %s header", ErrUnverified, giteaSignatureHeader) + } + got, err := hex.DecodeString(headerValue) + if err != nil { + return fmt.Errorf("%w: %s header is not valid hex", ErrUnverified, giteaSignatureHeader) + } + mac := hmac.New(sha256.New, []byte(secret)) + mac.Write(body) + if !hmac.Equal(got, mac.Sum(nil)) { + return fmt.Errorf("%w: %s does not match body", ErrUnverified, giteaSignatureHeader) + } + return nil +} + +// VerifySharedSecret constant-time compares the value of the provider's +// secret header against the configured secret. +func VerifySharedSecret(headerName, headerValue, secret string) error { + if headerValue == "" { + return fmt.Errorf("%w: missing %s header", ErrUnverified, headerName) + } + if subtle.ConstantTimeCompare([]byte(headerValue), []byte(secret)) != 1 { + return fmt.Errorf("%w: %s does not verify", ErrUnverified, headerName) + } + return nil +} + +// GiteaSignatureHeader exposes the fixed HMAC header name (config/docs). +func GiteaSignatureHeader() string { return giteaSignatureHeader } + +// SignGitea computes the expected HMAC for a body+secret (tests, smoke). +func SignGitea(body []byte, secret string) string { + mac := hmac.New(sha256.New, []byte(secret)) + mac.Write(body) + return hex.EncodeToString(mac.Sum(nil)) +} diff --git a/internal/events/verify_test.go b/internal/events/verify_test.go new file mode 100644 index 0000000..65058cc --- /dev/null +++ b/internal/events/verify_test.go @@ -0,0 +1,98 @@ +package events + +import ( + "crypto/hmac" + "crypto/sha256" + "encoding/hex" + "strings" + "testing" +) + +func hmacHex(body []byte, secret string) string { + mac := hmac.New(sha256.New, []byte(secret)) + mac.Write(body) + return hex.EncodeToString(mac.Sum(nil)) +} + +func TestVerifyGiteaHMAC(t *testing.T) { + body := []byte(`{"action":"approved","number":5}`) + secret := "gitea-hook-secret" + + cases := []struct { + name string + sig string + wantErr bool + }{ + {"valid signature", hmacHex(body, secret), false}, + {"missing header value", "", true}, + {"all-zero signature", strings.Repeat("00", 32), true}, + {"wrong secret", hmacHex(body, "other-secret"), true}, + {"not hex", "not-hex-at-all", true}, + {"truncated signature", hmacHex(body, secret)[:32], true}, + {"uppercase hex still valid", strings.ToUpper(hmacHex(body, secret)), false}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + err := VerifyGiteaHMAC(body, tc.sig, secret) + if tc.wantErr && err == nil { + t.Fatalf("expected error, got nil") + } + if !tc.wantErr && err != nil { + t.Fatalf("unexpected error: %v", err) + } + if err != nil && !IsUnverified(err) { + t.Errorf("error %v is not ErrUnverified", err) + } + }) + } +} + +func TestVerifyGiteaHMACBodySensitivity(t *testing.T) { + secret := "s" + sig := hmacHex([]byte(`{"a":1}`), secret) + if err := VerifyGiteaHMAC([]byte(`{"a":2}`), sig, secret); err == nil { + t.Fatal("signature over a different body must not verify") + } +} + +func TestVerifySharedSecret(t *testing.T) { + const header = "X-Test-Secret" + secret := "shared-hook-secret" + + cases := []struct { + name string + value string + wantErr bool + }{ + {"valid", secret, false}, + {"missing header value", "", true}, + {"wrong secret", "wrong", true}, + {"prefix of secret", secret[:10], true}, + {"case differs", strings.ToUpper(secret), true}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + err := VerifySharedSecret(header, tc.value, secret) + if tc.wantErr && err == nil { + t.Fatalf("expected error, got nil") + } + if !tc.wantErr && err != nil { + t.Fatalf("unexpected error: %v", err) + } + }) + } +} + +func TestSignGiteaMatchesVerify(t *testing.T) { + body, secret := []byte(`{}`), "k" + if err := VerifyGiteaHMAC(body, SignGitea(body, secret), secret); err != nil { + t.Fatalf("SignGitea does not verify: %v", err) + } + if GiteaSignatureHeader() != "X-Gitea-Signature" { + t.Errorf("unexpected header name %q", GiteaSignatureHeader()) + } + // sha256 import guard: helper must agree with a direct hmac. + if SignGitea(body, secret) != hmacHex(body, secret) { + t.Errorf("SignGitea disagrees with hmacHex") + } +}