diff --git a/flake.nix b/flake.nix index 12a58b7..28d17a1 100644 --- a/flake.nix +++ b/flake.nix @@ -43,7 +43,7 @@ "-s -w" ]; - vendorHash = "sha256-opS3f4GDczDRp7mrBzvQtK13Qi4snanX4I64FHTh7Pw="; + vendorHash = "sha256-LkHC/gFiSfYz9Z4bYMq1QNdapPYp8h1DSMRfFU9f7mw="; }; in { diff --git a/go.mod b/go.mod index 9e043ed..ee253e7 100644 --- a/go.mod +++ b/go.mod @@ -12,8 +12,8 @@ require ( github.com/getsentry/sentry-go/slog v0.40.0 github.com/go-git/go-billy/v6 v6.0.0-20251126203821-7f9c95185ee0 github.com/go-git/go-git/v6 v6.0.0-20251128074608-48f817f57805 - github.com/influxdata/influxdb v1.12.2 github.com/jpillora/backoff v1.0.0 + github.com/kankanreno/go-snowflake v1.2.0 github.com/klauspost/compress v1.18.1 github.com/maypok86/otter/v2 v2.2.1 github.com/minio/minio-go/v7 v7.0.97 diff --git a/go.sum b/go.sum index 52e0bd7..23c8742 100644 --- a/go.sum +++ b/go.sum @@ -57,10 +57,10 @@ github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= -github.com/influxdata/influxdb v1.12.2 h1:Y0ZBu47gYVbDCRPMFOrlRRZ3grdqPGIJxerFysVSq+g= -github.com/influxdata/influxdb v1.12.2/go.mod h1:EwqFMB6GKV0Huug82Msa5f8QfXhqETUmC4L9A0QZJQM= github.com/jpillora/backoff v1.0.0 h1:uvFg412JmmHBHw7iwprIxkPMI+sGQ4kzOWsMeHnm2EA= github.com/jpillora/backoff v1.0.0/go.mod h1:J/6gKK9jxlEcS3zixgDgUAsiuZ7yrSoa/FX5e0EB2j4= +github.com/kankanreno/go-snowflake v1.2.0 h1:Zx2SctsH5pivIj9vyhwyDyQS23jcDJx4iT49Bjv81kk= +github.com/kankanreno/go-snowflake v1.2.0/go.mod h1:6CZ+10PeVsFXKZUTYyJzPiRIjn1IXbInaWLCX/LDJ0g= github.com/kevinburke/ssh_config v1.4.0 h1:6xxtP5bZ2E4NF5tuQulISpTO2z8XbtH8cg1PWkxoFkQ= github.com/kevinburke/ssh_config v1.4.0/go.mod h1:q2RIzfka+BXARoNexmF9gkxEX7DmvbW9P4hIVx2Kg4M= github.com/klauspost/compress v1.18.1 h1:bcSGx7UbpBqMChDtsF28Lw6v/G94LPrrbMbdC3JH2co= diff --git a/src/audit.go b/src/audit.go index a7cdc31..4207a35 100644 --- a/src/audit.go +++ b/src/audit.go @@ -1,18 +1,61 @@ package git_pages import ( + "cmp" "context" "fmt" "net/http" + "strconv" "strings" "time" - "github.com/influxdata/influxdb/pkg/snowflake" exponential "github.com/jpillora/backoff" + "github.com/kankanreno/go-snowflake" + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/promauto" "google.golang.org/protobuf/proto" timestamppb "google.golang.org/protobuf/types/known/timestamppb" ) +var ( + auditNotifyOkCount = promauto.NewCounter(prometheus.CounterOpts{ + Name: "git_pages_audit_notify_ok", + Help: "Count of successful audit notifications", + }) + auditNotifyErrorCount = promauto.NewCounter(prometheus.CounterOpts{ + Name: "git_pages_audit_notify_error", + Help: "Count of failed audit notifications", + }) +) + +type AuditID int64 + +func GenerateAuditID() AuditID { + inner, err := snowflake.NextID() + if err != nil { + panic(err) + } + return AuditID(inner) +} + +func ParseAuditID(repr string) (AuditID, error) { + inner, err := strconv.ParseInt(repr, 16, 64) + if err != nil { + return AuditID(0), err + } + return AuditID(inner), nil +} + +func (id AuditID) String() string { + return fmt.Sprintf("%016x", int64(id)) +} + +func (id AuditID) CompareTime(when time.Time) int { + idMillis := int64(id) >> (snowflake.MachineIDLength + snowflake.SequenceLength) + whenMillis := when.UTC().UnixNano() / 1e6 + return cmp.Compare(idMillis, whenMillis) +} + func EncodeAuditRecord(auditRecord *AuditRecord) (data []byte) { data, err := proto.MarshalOptions{Deterministic: true}.Marshal(auditRecord) if err != nil { @@ -29,15 +72,13 @@ func DecodeAuditRecord(data []byte) (auditRecord *AuditRecord, err error) { type auditedBackend struct { Backend - ids *snowflake.Generator } var _ Backend = (*auditedBackend)(nil) func NewAuditedBackend(backend Backend) Backend { if config.Feature("audit") { - ids := snowflake.New(config.Audit.NodeID) - return &auditedBackend{backend, ids} + return &auditedBackend{backend} } else { return backend } @@ -50,11 +91,12 @@ func NewAuditedBackend(backend Backend) Backend { // to be a 100% accurate reflection of performed actions. When in doubt, the audit records // should be examined together with the application logs. func (audited *auditedBackend) appendNewAuditRecord(ctx context.Context, record *AuditRecord) (err error) { - record.Timestamp = timestamppb.Now() - if config.Audit.Collect { - id := fmt.Sprintf("%016x", audited.ids.Next()) - err = audited.Backend.AppendAuditRecord(ctx, id, record) + id := GenerateAuditID() + record.Id = proto.Int64(int64(id)) + record.Timestamp = timestamppb.Now() + + err = audited.Backend.AppendAuditLog(ctx, id, record) if err != nil { err = fmt.Errorf("audit: %w", err) } else { @@ -64,7 +106,7 @@ func (audited *auditedBackend) appendNewAuditRecord(ctx context.Context, record } else { subject = fmt.Sprintf("%s/%s", *record.Domain, *record.Project) } - logc.Printf(ctx, "audit %s ok: %s %s\n", subject, record.Event.String(), id) + logc.Printf(ctx, "audit %s ok: %s %s\n", subject, id, record.Event.String()) // Send a notification to the audit server, if configured, and try to make sure // it is delivered by retrying with exponential backoff on errors. @@ -74,10 +116,11 @@ func (audited *auditedBackend) appendNewAuditRecord(ctx context.Context, record return } -func notifyAudit(ctx context.Context, id string) { +func notifyAudit(ctx context.Context, id AuditID) { if config.Audit.NotifyURL != nil { notifyURL := config.Audit.NotifyURL.URL - notifyURL.RawQuery = id + notifyURL.RawQuery = id.String() + go func() { backoff := exponential.Backoff{ Jitter: true, @@ -89,9 +132,11 @@ func notifyAudit(ctx context.Context, id string) { if err != nil { sleepFor := backoff.Duration() logc.Printf(ctx, "audit notify %s err: %s (retry in %s)", id, err, sleepFor) + auditNotifyErrorCount.Inc() time.Sleep(sleepFor) } else { logc.Printf(ctx, "audit notify %s ok", id) + auditNotifyOkCount.Inc() break } } diff --git a/src/backend.go b/src/backend.go index d7cfdf9..54af8ea 100644 --- a/src/backend.go +++ b/src/backend.go @@ -5,6 +5,7 @@ import ( "errors" "fmt" "io" + "iter" "slices" "strings" "time" @@ -29,9 +30,27 @@ const ( ) type GetManifestOptions struct { + // If true and the manifest is past the cache `MaxAge`, `GetManifest` blocks and returns + // a fresh object instead of revalidating in background and returning a stale object. BypassCache bool } +type QueryAuditLogOptions struct { + // Inclusive lower bound on returned audit records, per their Snowflake ID (which may differ + // slightly from the embedded timestamp). If zero, audit records are returned since beginning + // of time. + Since time.Time + // Inclusive upper bound on returned audit records, per their Snowflake ID (which may differ + // slightly from the embedded timestamp). If zero, audit records are returned until the end + // of time. + Until time.Time +} + +type QueryAuditLogResult struct { + ID AuditID + Err error +} + type Backend interface { // Returns true if the feature has been enabled for this store, false otherwise. HasFeature(ctx context.Context, feature BackendFeature) bool @@ -82,8 +101,14 @@ type Backend interface { // is discovered serving abusive content. FreezeDomain(ctx context.Context, domain string, freeze bool) error - // Append an audit record to the log. - AppendAuditRecord(ctx context.Context, id string, record *AuditRecord) error + // Append a record to the audit log. + AppendAuditLog(ctx context.Context, id AuditID, record *AuditRecord) error + + // Retrieve a single record from the audit log. + QueryAuditLog(ctx context.Context, id AuditID) (record *AuditRecord, err error) + + // Retrieve records from the audit log by time range. + SearchAuditLog(ctx context.Context, opts QueryAuditLogOptions) iter.Seq[QueryAuditLogResult] } func CreateBackend(config *StorageConfig) (backend Backend, err error) { diff --git a/src/backend_fs.go b/src/backend_fs.go index 5f593b5..c94e348 100644 --- a/src/backend_fs.go +++ b/src/backend_fs.go @@ -6,7 +6,8 @@ import ( "errors" "fmt" "io" - "io/fs" + iofs "io/fs" + "iter" "os" "path/filepath" "strings" @@ -154,18 +155,19 @@ func (fs *FSBackend) DeleteBlob(ctx context.Context, name string) error { return fs.blobRoot.Remove(blobPath) } -func (b *FSBackend) ListManifests(ctx context.Context) (manifests []string, err error) { - err = fs.WalkDir(b.siteRoot.FS(), ".", func(path string, d fs.DirEntry, err error) error { - if strings.Count(path, "/") > 1 { - return fs.SkipDir - } - _, project, _ := strings.Cut(path, "/") - if project == "" || strings.HasPrefix(project, ".") && project != ".index" { +func (fs *FSBackend) ListManifests(ctx context.Context) (manifests []string, err error) { + err = iofs.WalkDir(fs.siteRoot.FS(), ".", + func(path string, entry iofs.DirEntry, err error) error { + if strings.Count(path, "/") > 1 { + return iofs.SkipDir + } + _, project, _ := strings.Cut(path, "/") + if project == "" || strings.HasPrefix(project, ".") && project != ".index" { + return nil + } + manifests = append(manifests, path) return nil - } - manifests = append(manifests, path) - return nil - }) + }) return } @@ -293,10 +295,50 @@ func (fs *FSBackend) FreezeDomain(ctx context.Context, domain string, freeze boo } } -func (fs *FSBackend) AppendAuditRecord(ctx context.Context, id string, record *AuditRecord) error { - if _, err := fs.auditRoot.Stat(id); err == nil { +func (fs *FSBackend) AppendAuditLog(ctx context.Context, id AuditID, record *AuditRecord) error { + if _, err := fs.auditRoot.Stat(id.String()); err == nil { panic(fmt.Errorf("audit ID collision: %s", id)) } - return fs.auditRoot.WriteFile(id, EncodeAuditRecord(record), 0o644) + return fs.auditRoot.WriteFile(id.String(), EncodeAuditRecord(record), 0o644) +} + +func (fs *FSBackend) QueryAuditLog(ctx context.Context, id AuditID) (*AuditRecord, error) { + if data, err := fs.auditRoot.ReadFile(id.String()); err != nil { + return nil, fmt.Errorf("read: %w", err) + } else if record, err := DecodeAuditRecord(data); err != nil { + return nil, fmt.Errorf("decode: %w", err) + } else { + return record, nil + } +} + +func (fs *FSBackend) SearchAuditLog( + ctx context.Context, opts QueryAuditLogOptions, +) iter.Seq[QueryAuditLogResult] { + return func(yield func(QueryAuditLogResult) bool) { + iofs.WalkDir(fs.auditRoot.FS(), ".", + func(path string, entry iofs.DirEntry, err error) error { + if path == "." { + return nil + } + var result QueryAuditLogResult + if err != nil { + result.Err = err + } else if id, err := ParseAuditID(path); err != nil { + result.Err = err + } else if !opts.Since.IsZero() && id.CompareTime(opts.Since) < 0 { + return nil + } else if !opts.Until.IsZero() && id.CompareTime(opts.Until) > 0 { + return nil + } else { + result.ID = id + } + if !yield(result) { + return iofs.SkipAll + } else { + return nil + } + }) + } } diff --git a/src/backend_s3.go b/src/backend_s3.go index 9c608a6..5880df3 100644 --- a/src/backend_s3.go +++ b/src/backend_s3.go @@ -6,6 +6,7 @@ import ( "crypto/sha256" "fmt" "io" + "iter" "net/http" "path" "strings" @@ -631,11 +632,13 @@ func (s3 *S3Backend) FreezeDomain(ctx context.Context, domain string, freeze boo } } -func auditObjectName(id string) string { +func auditObjectName(id AuditID) string { return fmt.Sprintf("audit/%s", id) } -func (s3 *S3Backend) AppendAuditRecord(ctx context.Context, id string, record *AuditRecord) error { +func (s3 *S3Backend) AppendAuditLog(ctx context.Context, id AuditID, record *AuditRecord) error { + logc.Printf(ctx, "s3: append audit %s\n", id) + name := auditObjectName(id) data := EncodeAuditRecord(record) @@ -648,3 +651,49 @@ func (s3 *S3Backend) AppendAuditRecord(ctx context.Context, id string, record *A } return err } + +func (s3 *S3Backend) QueryAuditLog(ctx context.Context, id AuditID) (*AuditRecord, error) { + logc.Printf(ctx, "s3: read audit %s\n", id) + + object, err := s3.client.GetObject(ctx, s3.bucket, auditObjectName(id), + minio.GetObjectOptions{}) + if err != nil { + return nil, err + } + defer object.Close() + + data, err := io.ReadAll(object) + if err != nil { + return nil, err + } + + return DecodeAuditRecord(data) +} + +func (s3 *S3Backend) SearchAuditLog( + ctx context.Context, opts QueryAuditLogOptions, +) iter.Seq[QueryAuditLogResult] { + return func(yield func(QueryAuditLogResult) bool) { + logc.Printf(ctx, "s3: query audit\n") + + ctx, cancel := context.WithCancel(ctx) + defer cancel() + + prefix := "audit/" + for object := range s3.client.ListObjectsIter(ctx, s3.bucket, minio.ListObjectsOptions{ + Prefix: prefix, + }) { + var result QueryAuditLogResult + if object.Err != nil { + result.Err = object.Err + } else if id, err := ParseAuditID(strings.TrimPrefix(object.Key, prefix)); err != nil { + result.Err = err + } else { + result.ID = id + } + if !yield(result) { + break + } + } + } +} diff --git a/src/config.go b/src/config.go index 29642f7..8b30ff1 100644 --- a/src/config.go +++ b/src/config.go @@ -148,7 +148,7 @@ type LimitsConfig struct { } type AuditConfig struct { - // Globally unique node identifier (0 to 1023 inclusive). + // Globally unique machine identifier (0 to 63 inclusive). NodeID int `toml:"node-id"` // Whether audit reports should be stored whenever an audit event occurs. Collect bool `toml:"collect"` diff --git a/src/main.go b/src/main.go index 6744184..5ac2d48 100644 --- a/src/main.go +++ b/src/main.go @@ -16,9 +16,11 @@ import ( "os" "runtime/debug" "strings" + "time" automemlimit "github.com/KimMachineGun/automemlimit/memlimit" "github.com/c2h5oh/datasize" + "github.com/kankanreno/go-snowflake" "github.com/prometheus/client_golang/prometheus/promhttp" ) @@ -85,6 +87,13 @@ func configureFallback(_ context.Context) (err error) { return } +// Thread-unsafe, must be called only during initial configuration. +func configureAudit(_ context.Context) (err error) { + snowflake.SetStartTime(time.Date(2025, 12, 1, 0, 0, 0, 0, time.UTC)) + snowflake.SetMachineID(config.Audit.NodeID) + return +} + func listen(ctx context.Context, name string, listen string) net.Listener { if listen == "-" { return nil @@ -256,6 +265,7 @@ func Main() { configureMemLimit(ctx), configureWildcards(ctx), configureFallback(ctx), + configureAudit(ctx), ); err != nil { logc.Fatalln(ctx, err) } diff --git a/src/observe.go b/src/observe.go index 30127fb..991e9d7 100644 --- a/src/observe.go +++ b/src/observe.go @@ -5,6 +5,7 @@ import ( "errors" "fmt" "io" + "iter" "log" "log/slog" "math/rand/v2" @@ -437,9 +438,33 @@ func (backend *observedBackend) FreezeDomain(ctx context.Context, domain string, return } -func (backend *observedBackend) AppendAuditRecord(ctx context.Context, id string, record *AuditRecord) (err error) { - span, ctx := ObserveFunction(ctx, "AppendAudit", "audit.id", id) - err = backend.inner.AppendAuditRecord(ctx, id, record) +func (backend *observedBackend) AppendAuditLog(ctx context.Context, id AuditID, record *AuditRecord) (err error) { + span, ctx := ObserveFunction(ctx, "AppendAuditLog", "audit.id", id) + err = backend.inner.AppendAuditLog(ctx, id, record) span.Finish() return } + +func (backend *observedBackend) QueryAuditLog(ctx context.Context, id AuditID) (record *AuditRecord, err error) { + span, ctx := ObserveFunction(ctx, "QueryAuditLog", "audit.id", id) + record, err = backend.inner.QueryAuditLog(ctx, id) + span.Finish() + return +} + +func (backend *observedBackend) SearchAuditLog( + ctx context.Context, opts QueryAuditLogOptions, +) iter.Seq[QueryAuditLogResult] { + return func(yield func(QueryAuditLogResult) bool) { + span, ctx := ObserveFunction(ctx, "SearchAuditLog", + "audit.search.since", opts.Since, + "audit.search.until", opts.Until, + ) + for result := range backend.inner.SearchAuditLog(ctx, opts) { + if !yield(result) { + break + } + } + span.Finish() + } +} diff --git a/src/schema.pb.go b/src/schema.pb.go index cfe3930..093d1e9 100644 --- a/src/schema.pb.go +++ b/src/schema.pb.go @@ -654,8 +654,9 @@ func (x *Manifest) GetProblems() []*Problem { type AuditRecord struct { state protoimpl.MessageState `protogen:"open.v1"` // Audit event metadata. - Event *AuditEvent `protobuf:"varint,1,opt,name=event,enum=AuditEvent" json:"event,omitempty"` + Id *int64 `protobuf:"varint,1,opt,name=id" json:"id,omitempty"` Timestamp *timestamppb.Timestamp `protobuf:"bytes,2,opt,name=timestamp" json:"timestamp,omitempty"` + Event *AuditEvent `protobuf:"varint,3,opt,name=event,enum=AuditEvent" json:"event,omitempty"` // Affected resource. Domain *string `protobuf:"bytes,10,opt,name=domain" json:"domain,omitempty"` Project *string `protobuf:"bytes,11,opt,name=project" json:"project,omitempty"` // only for `*Manifest` events @@ -695,11 +696,11 @@ func (*AuditRecord) Descriptor() ([]byte, []int) { return file_schema_proto_rawDescGZIP(), []int{6} } -func (x *AuditRecord) GetEvent() AuditEvent { - if x != nil && x.Event != nil { - return *x.Event +func (x *AuditRecord) GetId() int64 { + if x != nil && x.Id != nil { + return *x.Id } - return AuditEvent_InvalidEvent + return 0 } func (x *AuditRecord) GetTimestamp() *timestamppb.Timestamp { @@ -709,6 +710,13 @@ func (x *AuditRecord) GetTimestamp() *timestamppb.Timestamp { return nil } +func (x *AuditRecord) GetEvent() AuditEvent { + if x != nil && x.Event != nil { + return *x.Event + } + return AuditEvent_InvalidEvent +} + func (x *AuditRecord) GetDomain() string { if x != nil && x.Domain != nil { return *x.Domain @@ -775,10 +783,11 @@ const file_schema_proto_rawDesc = "" + "\bproblems\x18\a \x03(\v2\b.ProblemR\bproblems\x1aC\n" + "\rContentsEntry\x12\x10\n" + "\x03key\x18\x01 \x01(\tR\x03key\x12\x1c\n" + - "\x05value\x18\x02 \x01(\v2\x06.EntryR\x05value:\x028\x01\"\xc3\x01\n" + - "\vAuditRecord\x12!\n" + - "\x05event\x18\x01 \x01(\x0e2\v.AuditEventR\x05event\x128\n" + - "\ttimestamp\x18\x02 \x01(\v2\x1a.google.protobuf.TimestampR\ttimestamp\x12\x16\n" + + "\x05value\x18\x02 \x01(\v2\x06.EntryR\x05value:\x028\x01\"\xd3\x01\n" + + "\vAuditRecord\x12\x0e\n" + + "\x02id\x18\x01 \x01(\x03R\x02id\x128\n" + + "\ttimestamp\x18\x02 \x01(\v2\x1a.google.protobuf.TimestampR\ttimestamp\x12!\n" + + "\x05event\x18\x03 \x01(\x0e2\v.AuditEventR\x05event\x12\x16\n" + "\x06domain\x18\n" + " \x01(\tR\x06domain\x12\x18\n" + "\aproject\x18\v \x01(\tR\aproject\x12%\n" + @@ -837,8 +846,8 @@ var file_schema_proto_depIdxs = []int32{ 4, // 4: Manifest.redirects:type_name -> RedirectRule 6, // 5: Manifest.headers:type_name -> HeaderRule 7, // 6: Manifest.problems:type_name -> Problem - 2, // 7: AuditRecord.event:type_name -> AuditEvent - 11, // 8: AuditRecord.timestamp:type_name -> google.protobuf.Timestamp + 11, // 7: AuditRecord.timestamp:type_name -> google.protobuf.Timestamp + 2, // 8: AuditRecord.event:type_name -> AuditEvent 8, // 9: AuditRecord.manifest:type_name -> Manifest 3, // 10: Manifest.ContentsEntry.value:type_name -> Entry 11, // [11:11] is the sub-list for method output_type diff --git a/src/schema.proto b/src/schema.proto index 4ed4720..18a9bb7 100644 --- a/src/schema.proto +++ b/src/schema.proto @@ -116,8 +116,9 @@ enum AuditEvent { message AuditRecord { // Audit event metadata. - AuditEvent event = 1; + int64 id = 1; google.protobuf.Timestamp timestamp = 2; + AuditEvent event = 3; // Affected resource. string domain = 10;