ProofedMesh/services/attest-reconciler/internal/attest/attest.go
CTO 2e44147a71
All checks were successful
validate / JSON Schema (ajv) (push) Successful in 19s
validate / Go (vet + test) (push) Successful in 19s
validate / Rego (opa fmt + opa test) (push) Successful in 14s
services: add attest-reconciler Go service (periodic worker re-signs per_diem_rates rows without rate_change_attestation_ref via Signer+TLog, 11/11 go tests pass, go vet clean)
- services/attest-reconciler/ as its own go.mod (admin-api internal/ is not importable; shared code gets extracted when a 3rd consumer appears).
- Scheduler accepts "@every <dur>" or a bare Go duration; context-cancellation stops the loop cleanly.
- Store client defines PendingRateStore + Postgres skeleton for SELECT ... WHERE rate_change_attestation_ref IS NULL.
- Orchestrator + HTTPSigner + tlog.HTTPSubmitter mirror admin-api byte-for-byte so the TLog idempotency key collides on resubmits (kein Doppelsignieren).
- Prometheus text-format /metrics endpoint, dependency-free: attest_reconciler_dirty_rows_total, _reconciliation_attempts_total, _reconciliation_failures_total, _idempotent_noop_total.
- Tests with fake store + fake signer + fake submitter cover: happy path, already-attested rows are NOT re-signed, TLog idempotency across ticks, TLog-down failure path, pre-cancelled ctx.
- Top-level README reflects the new service; CI auto-discovers services/*/go.mod so no workflow change needed.

Co-Authored-By: Paperclip <noreply@paperclip.ing>
2026-10-25 18:06:36 +00:00

238 lines
7.5 KiB
Go

// Package attest mirrors the admin-api attestation API (ADR-0005). Current
// state: the reconciler re-declares the small Signer/Submitter contract so it
// can live in its own go.mod without pulling admin-api's internal/ tree.
// When a third consumer appears (per the parent issue's note), these types
// move into a shared module and both admin-api and attest-reconciler switch
// their imports; the wire types do not change.
package attest
import (
"context"
"encoding/base64"
"encoding/json"
"errors"
"fmt"
"time"
)
// Statement is the in-toto v1 Statement shape handed to the Signing Service.
type Statement struct {
Type string `json:"_type"`
Subject []Subject `json:"subject"`
PredicateType string `json:"predicateType"`
Predicate map[string]any `json:"predicate"`
}
type Subject struct {
Name string `json:"name"`
Digest map[string]string `json:"digest"`
}
type Envelope struct {
PayloadType string `json:"payloadType"`
Payload string `json:"payload"` // base64url
Signatures []EnvelopeSignature `json:"signatures"`
}
type EnvelopeSignature struct {
KeyID string `json:"keyid"`
Sig string `json:"sig"`
}
type LogReceipt struct {
LogIndex int64 `json:"log_index"`
ContentHash string `json:"content_hash"`
LoggedAt time.Time `json:"logged_at"`
InclusionProof InclusionProof `json:"inclusion_proof"`
}
type InclusionProof struct {
TreeSize int64 `json:"tree_size"`
TreeRootHash string `json:"tree_root_hash"`
SignedTreeHeadRef string `json:"signed_tree_head_ref"`
Hashes []string `json:"hashes"`
}
// Signer wraps POST /api/v1/sign/dsse.
type Signer interface {
SignDSSE(ctx context.Context, keyRef string, stmt Statement) (envelope Envelope, contentHash string, err error)
}
// Submitter wraps POST /api/v1/log/entries. MUST be idempotent on
// (envelope, contentHash); that is the whole reason the reconciler can safely
// re-run already-attested rows without producing duplicates.
type Submitter interface {
Submit(ctx context.Context, env Envelope, contentHash string) (LogReceipt, error)
}
type TokenSource interface {
Token(ctx context.Context) (string, error)
}
type StaticToken string
func (s StaticToken) Token(_ context.Context) (string, error) { return string(s), nil }
// ChangeKind labels why a rate changed.
type ChangeKind string
const (
ChangeCreate ChangeKind = "create"
ChangeRetire ChangeKind = "retire"
)
// Rate is the subset of per_diem_rates the reconciler needs to re-sign a row.
// Field layout mirrors admin-api/internal/model.Rate so JSON round-trips with
// the same canonical bytes.
type Rate struct {
RateID string `json:"rate_id"`
Country string `json:"country"`
City *string `json:"city,omitempty"`
Currency string `json:"currency"`
FullDay string `json:"full_day"`
PartialDayMin8h string `json:"partial_day_min8h"`
ArrivalDay string `json:"arrival_day"`
DepartureDay string `json:"departure_day"`
ValidFrom time.Time `json:"valid_from"`
ValidTo *time.Time `json:"valid_to,omitempty"`
SourceRegulation string `json:"source_regulation"`
CreatedBy string `json:"created_by"`
CreatedAt time.Time `json:"created_at"`
}
// RateChangeEvent is what the reconciler reconstructs for a pending row.
type RateChangeEvent struct {
Kind ChangeKind
Rate Rate
PerformedBy PerformedBy
PerformedAt time.Time
}
type PerformedBy struct {
Issuer string `json:"iss"`
Subject string `json:"sub"`
SessionID string `json:"session_id"`
}
// Orchestrator signs a rate_change event, submits it to the TLog, and returns
// the attestation content_hash. Semantics match admin-api exactly: the TLog
// side is idempotent, so re-running an already-signed event is safe and
// returns the same content_hash.
type Orchestrator struct {
Signer Signer
Submitter Submitter
KeyRef string
}
func (o Orchestrator) SignAndLog(ctx context.Context, ev RateChangeEvent) (string, LogReceipt, error) {
if o.Signer == nil {
return "", LogReceipt{}, errors.New("attest.Orchestrator: Signer is nil")
}
if o.Submitter == nil {
return "", LogReceipt{}, errors.New("attest.Orchestrator: Submitter is nil")
}
if o.KeyRef == "" {
return "", LogReceipt{}, errors.New("attest.Orchestrator: KeyRef is empty")
}
stmt, err := BuildStatement(ev)
if err != nil {
return "", LogReceipt{}, fmt.Errorf("build statement: %w", err)
}
env, contentHash, err := o.Signer.SignDSSE(ctx, o.KeyRef, stmt)
if err != nil {
return "", LogReceipt{}, fmt.Errorf("sign dsse: %w", err)
}
receipt, err := o.Submitter.Submit(ctx, env, contentHash)
if err != nil {
return contentHash, LogReceipt{}, fmt.Errorf("submit tlog: %w", err)
}
if receipt.ContentHash != contentHash {
return contentHash, receipt, fmt.Errorf("tlog returned content_hash %q, expected %q", receipt.ContentHash, contentHash)
}
return contentHash, receipt, nil
}
// BuildStatement produces the canonical in-toto v1 Statement for a rate change.
// Byte-identical to admin-api/internal/attest.buildStatement so the Signing
// Service receives the same canonical bytes for the same row and the TLog
// idempotency key matches on both sides.
func BuildStatement(ev RateChangeEvent) (Statement, error) {
rateHash, err := hashRateCore(ev.Rate)
if err != nil {
return Statement{}, err
}
predicate := map[string]any{
"change_kind": string(ev.Kind),
"rate_id": ev.Rate.RateID,
"country": ev.Rate.Country,
"city": ev.Rate.City,
"currency": ev.Rate.Currency,
"full_day": ev.Rate.FullDay,
"partial_day_min8h": ev.Rate.PartialDayMin8h,
"arrival_day": ev.Rate.ArrivalDay,
"departure_day": ev.Rate.DepartureDay,
"valid_from": ev.Rate.ValidFrom.Format("2006-01-02"),
"valid_to": nilDate(ev.Rate.ValidTo),
"source_regulation": ev.Rate.SourceRegulation,
"performed_by": ev.PerformedBy,
"performed_at": ev.PerformedAt.UTC().Format(time.RFC3339),
}
return Statement{
Type: "https://in-toto.io/Statement/v1",
Subject: []Subject{{
Name: fmt.Sprintf("per_diem_rate/%s", ev.Rate.RateID),
Digest: map[string]string{"sha256": rateHash},
}},
PredicateType: "https://basicstack.de/attestations/rate_change/v1",
Predicate: predicate,
}, nil
}
func nilDate(t *time.Time) any {
if t == nil {
return nil
}
return t.Format("2006-01-02")
}
func hashRateCore(r Rate) (string, error) {
core := map[string]any{
"rate_id": r.RateID,
"country": r.Country,
"city": r.City,
"currency": r.Currency,
"full_day": r.FullDay,
"partial_day_min8h": r.PartialDayMin8h,
"arrival_day": r.ArrivalDay,
"departure_day": r.DepartureDay,
"valid_from": r.ValidFrom.Format("2006-01-02"),
"source_regulation": r.SourceRegulation,
}
b, err := canonicalJSON(core)
if err != nil {
return "", err
}
return "sha256:" + sha256hex(b), nil
}
var base64urlStd = base64.RawURLEncoding
// EncodePayload is kept for parity with admin-api; the reconciler uses it only
// in tests that construct an envelope directly.
func EncodePayload(stmt Statement) (string, []byte, error) {
b, err := canonicalJSON(stmt)
if err != nil {
return "", nil, err
}
return base64urlStd.EncodeToString(b), b, nil
}
func EnvelopeContentHash(env Envelope) (string, error) {
b, err := canonicalJSON(env)
if err != nil {
return "", err
}
return "sha256:" + sha256hex(b), nil
}
var _ = json.Marshal