From 24d831ca9bb1db05d190ee198d218e5981ff7250 Mon Sep 17 00:00:00 2001 From: Manish <69393822+manish-9245@users.noreply.github.com> Date: Sat, 24 Oct 2026 12:05:11 +0530 Subject: [PATCH 01/54] feat(server): single-binary skeleton with three listeners (#87) * feat(server): single-binary skeleton with three listeners Public, ingest and admin listeners with graceful shutdown, JSON config with env overrides, loopback-only admin, /healthz, subcommand stubs and a static cross-platform build matrix. * fix: reject serve args, surface listener failures, shut down listeners concurrently --- .github/workflows/go.yml | 25 ++++++ .gitignore | 5 ++ Makefile | 10 +++ cmd/kaasu/main.go | 83 ++++++++++++++++++++ cmd/kaasu/main_test.go | 38 +++++++++ go.mod | 3 + internal/config/config.go | 85 ++++++++++++++++++++ internal/config/config_test.go | 82 ++++++++++++++++++++ internal/server/server.go | 137 +++++++++++++++++++++++++++++++++ internal/server/server_test.go | 86 +++++++++++++++++++++ 10 files changed, 554 insertions(+) create mode 100644 .github/workflows/go.yml create mode 100644 .gitignore create mode 100644 Makefile create mode 100644 cmd/kaasu/main.go create mode 100644 cmd/kaasu/main_test.go create mode 100644 go.mod create mode 100644 internal/config/config.go create mode 100644 internal/config/config_test.go create mode 100644 internal/server/server.go create mode 100644 internal/server/server_test.go diff --git a/.github/workflows/go.yml b/.github/workflows/go.yml new file mode 100644 index 0000000..f731e23 --- /dev/null +++ b/.github/workflows/go.yml @@ -0,0 +1,25 @@ +name: go +on: + pull_request: + push: + branches: [main] +permissions: + contents: read +jobs: + build: + runs-on: ubuntu-latest + strategy: + matrix: + include: + - { goos: linux, goarch: amd64 } + - { goos: linux, goarch: arm64 } + - { goos: darwin, goarch: amd64 } + - { goos: darwin, goarch: arm64 } + - { goos: windows, goarch: amd64 } + - { goos: windows, goarch: arm64 } + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-go@v5 + with: { go-version-file: go.mod } + - run: go build -trimpath -ldflags "-s -w" -o kaasu-${{ matrix.goos }}-${{ matrix.goarch }} ./cmd/kaasu + env: { CGO_ENABLED: '0', GOOS: '${{ matrix.goos }}', GOARCH: '${{ matrix.goarch }}' } diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..2bf38d4 --- /dev/null +++ b/.gitignore @@ -0,0 +1,5 @@ +bin/ +*.db +*.db-wal +*.db-shm +.agents/plans/ diff --git a/Makefile b/Makefile new file mode 100644 index 0000000..90a5692 --- /dev/null +++ b/Makefile @@ -0,0 +1,10 @@ +.PHONY: dev test lint build +dev: + go run ./cmd/kaasu serve +test: + go test -race ./... +lint: + @test -z "$$(gofmt -l .)" || (gofmt -l . && exit 1) + go vet ./... +build: + CGO_ENABLED=0 go build -trimpath -ldflags "-s -w" -o bin/kaasu ./cmd/kaasu diff --git a/cmd/kaasu/main.go b/cmd/kaasu/main.go new file mode 100644 index 0000000..5988a0a --- /dev/null +++ b/cmd/kaasu/main.go @@ -0,0 +1,83 @@ +// Command kaasu is the single Kaasu binary: server, tools and simulator. +package main + +import ( + "context" + "fmt" + "io" + "log/slog" + "os" + "os/signal" + "syscall" + "time" + + "github.com/opensawce/kaasu/internal/config" + "github.com/opensawce/kaasu/internal/server" +) + +// version is set at build time with -ldflags "-X main.version=...". +var version = "dev" + +const usage = "usage: kaasu [flags]" + +func main() { + os.Exit(run(os.Args[1:], os.Getenv, os.Stdout, os.Stderr)) +} + +func run(args []string, getenv func(string) string, stdout, stderr io.Writer) int { + if len(args) == 0 { + fmt.Fprintln(stderr, usage) + return 2 + } + switch args[0] { + case "version": + fmt.Fprintf(stdout, "kaasu %s\n", version) + return 0 + case "serve": + return serve(args[1:], getenv, stderr) + case "migrate", "verify", "backup", "pair": + fmt.Fprintf(stderr, "kaasu %s: not implemented yet\n", args[0]) + return 1 + default: + fmt.Fprintf(stderr, "unknown command %q\n%s\n", args[0], usage) + return 2 + } +} + +func serve(args []string, getenv func(string) string, stderr io.Writer) int { + if len(args) > 0 { + fmt.Fprintf(stderr, "serve: unexpected arguments %q (configure with KAASU_CONFIG or KAASU_* env)\n", args) + return 2 + } + cfg, err := config.Load(getenv, getenv("KAASU_CONFIG")) + if err != nil { + fmt.Fprintln(stderr, "config:", err) + return 1 + } + var level slog.Level + _ = level.UnmarshalText([]byte(cfg.LogLevel)) + log := slog.New(slog.NewJSONHandler(stderr, &slog.HandlerOptions{Level: level})) + + s := server.New(cfg, log) + if err := s.Start(); err != nil { + log.Error("start", "err", err) + return 1 + } + ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) + defer stop() + code := 0 + select { + case <-ctx.Done(): + case err := <-s.Err(): + log.Error("listener failed", "err", err) + code = 1 + } + log.Info("shutting down") + shutdownCtx, cancel := context.WithTimeout(context.Background(), 15*time.Second) + defer cancel() + if err := s.Shutdown(shutdownCtx); err != nil { + log.Error("shutdown", "err", err) + return 1 + } + return code +} diff --git a/cmd/kaasu/main_test.go b/cmd/kaasu/main_test.go new file mode 100644 index 0000000..0581a46 --- /dev/null +++ b/cmd/kaasu/main_test.go @@ -0,0 +1,38 @@ +package main + +import ( + "bytes" + "strings" + "testing" +) + +func TestRun(t *testing.T) { + tests := []struct { + name string + args []string + wantCode int + wantOut string + wantErr string + }{ + {"version", []string{"version"}, 0, "kaasu ", ""}, + {"no command", nil, 2, "", "usage:"}, + {"unknown command", []string{"frobnicate"}, 2, "", "unknown command"}, + {"stub pair", []string{"pair"}, 1, "", "not implemented"}, + {"stub backup", []string{"backup"}, 1, "", "not implemented"}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + var out, errb bytes.Buffer + code := run(tt.args, func(string) string { return "" }, &out, &errb) + if code != tt.wantCode { + t.Errorf("code=%d want %d (stderr=%q)", code, tt.wantCode, errb.String()) + } + if !strings.Contains(out.String(), tt.wantOut) { + t.Errorf("stdout=%q want %q", out.String(), tt.wantOut) + } + if !strings.Contains(errb.String(), tt.wantErr) { + t.Errorf("stderr=%q want %q", errb.String(), tt.wantErr) + } + }) + } +} diff --git a/go.mod b/go.mod new file mode 100644 index 0000000..76b25c6 --- /dev/null +++ b/go.mod @@ -0,0 +1,3 @@ +module github.com/opensawce/kaasu + +go 1.25 diff --git a/internal/config/config.go b/internal/config/config.go new file mode 100644 index 0000000..d187bb8 --- /dev/null +++ b/internal/config/config.go @@ -0,0 +1,85 @@ +// Package config loads Kaasu settings from an optional JSON file and KAASU_* environment variables. +package config + +import ( + "bytes" + "encoding/json" + "fmt" + "net" + "os" + "strings" +) + +// Config holds process-wide settings. Money and secrets never live here. +type Config struct { + PublicAddr string `json:"public_addr"` + IngestAddr string `json:"ingest_addr"` + AdminAddr string `json:"admin_addr"` + AdminAllowNonLoopback bool `json:"admin_allow_non_loopback"` + DatabasePath string `json:"database_path"` + LogLevel string `json:"log_level"` +} + +func defaults() Config { + return Config{ + PublicAddr: ":8080", // :443 + TLS is deferred to the deploy card; set KAASU_PUBLIC_ADDR behind a proxy + IngestAddr: ":8443", + AdminAddr: "127.0.0.1:9000", + DatabasePath: "kaasu.db", + LogLevel: "info", + } +} + +// Load reads path (if not empty), then lets KAASU_* variables override it, then validates. +func Load(getenv func(string) string, path string) (Config, error) { + c := defaults() + if path != "" { + b, err := os.ReadFile(path) + if err != nil { + return Config{}, fmt.Errorf("read config file: %w", err) + } + dec := json.NewDecoder(bytes.NewReader(b)) + dec.DisallowUnknownFields() + if err := dec.Decode(&c); err != nil { + return Config{}, fmt.Errorf("parse config file %s: %w", path, err) + } + } + str := func(dst *string, key string) { + if v := getenv(key); v != "" { + *dst = v + } + } + str(&c.PublicAddr, "KAASU_PUBLIC_ADDR") + str(&c.IngestAddr, "KAASU_INGEST_ADDR") + str(&c.AdminAddr, "KAASU_ADMIN_ADDR") + str(&c.DatabasePath, "KAASU_DATABASE_PATH") + str(&c.LogLevel, "KAASU_LOG_LEVEL") + if v := getenv("KAASU_ADMIN_ALLOW_NON_LOOPBACK"); v != "" { + c.AdminAllowNonLoopback = v == "1" || strings.EqualFold(v, "true") + } + return c, c.validate() +} + +func (c Config) validate() error { + switch c.LogLevel { + case "debug", "info", "warn", "error": + default: + return fmt.Errorf("log level %q: want debug, info, warn or error", c.LogLevel) + } + if !c.AdminAllowNonLoopback && !isLoopback(c.AdminAddr) { + return fmt.Errorf("admin address %q is not loopback; set admin_allow_non_loopback only behind a VPN", c.AdminAddr) + } + return nil +} + +func isLoopback(addr string) bool { + host, _, err := net.SplitHostPort(addr) + if err != nil || host == "" { + return false + } + if host == "localhost" { + return true + } + ip := net.ParseIP(host) + return ip != nil && ip.IsLoopback() +} diff --git a/internal/config/config_test.go b/internal/config/config_test.go new file mode 100644 index 0000000..f88f6a0 --- /dev/null +++ b/internal/config/config_test.go @@ -0,0 +1,82 @@ +package config + +import ( + "os" + "path/filepath" + "strings" + "testing" +) + +func env(m map[string]string) func(string) string { + return func(k string) string { return m[k] } +} + +func TestDefaults(t *testing.T) { + c, err := Load(env(nil), "") + if err != nil { + t.Fatal(err) + } + if c.PublicAddr != ":8080" || c.IngestAddr != ":8443" || c.AdminAddr != "127.0.0.1:9000" { + t.Fatalf("unexpected defaults: %+v", c) + } + if c.DatabasePath != "kaasu.db" || c.LogLevel != "info" { + t.Fatalf("unexpected defaults: %+v", c) + } +} + +func TestEnvOverridesFile(t *testing.T) { + path := filepath.Join(t.TempDir(), "kaasu.json") + if err := os.WriteFile(path, []byte(`{"public_addr":":9090","database_path":"from-file.db"}`), 0o600); err != nil { + t.Fatal(err) + } + c, err := Load(env(map[string]string{"KAASU_PUBLIC_ADDR": ":7070"}), path) + if err != nil { + t.Fatal(err) + } + if c.PublicAddr != ":7070" { + t.Errorf("env should beat file, got %q", c.PublicAddr) + } + if c.DatabasePath != "from-file.db" { + t.Errorf("file value lost, got %q", c.DatabasePath) + } +} + +func TestAdminMustBeLoopback(t *testing.T) { + tests := []struct { + name string + addr string + allow string + wantErr bool + }{ + {"loopback v4", "127.0.0.1:9000", "", false}, + {"loopback v6", "[::1]:9000", "", false}, + {"localhost", "localhost:9000", "", false}, + {"all interfaces", ":9000", "", true}, + {"public ip", "203.0.113.5:9000", "", true}, + {"vpn ip with explicit allow", "10.8.0.2:9000", "1", false}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + _, err := Load(env(map[string]string{ + "KAASU_ADMIN_ADDR": tt.addr, + "KAASU_ADMIN_ALLOW_NON_LOOPBACK": tt.allow, + }), "") + if (err != nil) != tt.wantErr { + t.Fatalf("err=%v wantErr=%v", err, tt.wantErr) + } + }) + } +} + +func TestRejectsUnknownFileKeysAndBadLevel(t *testing.T) { + path := filepath.Join(t.TempDir(), "kaasu.json") + if err := os.WriteFile(path, []byte(`{"public_adr":":1"}`), 0o600); err != nil { + t.Fatal(err) + } + if _, err := Load(env(nil), path); err == nil || !strings.Contains(err.Error(), "public_adr") { + t.Errorf("typo in config file should be reported, got %v", err) + } + if _, err := Load(env(map[string]string{"KAASU_LOG_LEVEL": "shout"}), ""); err == nil { + t.Error("bad log level should fail") + } +} diff --git a/internal/server/server.go b/internal/server/server.go new file mode 100644 index 0000000..69f8f93 --- /dev/null +++ b/internal/server/server.go @@ -0,0 +1,137 @@ +// Package server runs the three Kaasu listeners: public, ingest and admin. +// Admin routes are mounted only on the admin mux, so they cannot leak onto the other two. +package server + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "log/slog" + "net" + "net/http" + "sync" + "time" + + "github.com/opensawce/kaasu/internal/config" +) + +// Addrs are the bound listener addresses (useful when configured with port 0). +type Addrs struct{ Public, Ingest, Admin string } + +type listener struct { + name string + srv *http.Server + ln net.Listener +} + +// Server owns the three HTTP servers. +type Server struct { + cfg config.Config + log *slog.Logger + ls []*listener + // errc reports a listener that stopped for any reason other than Shutdown. + errc chan error +} + +func New(cfg config.Config, log *slog.Logger) *Server { + return &Server{cfg: cfg, log: log, errc: make(chan error, 3)} +} + +func healthz(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(map[string]string{"status": "ok"}) +} + +func baseMux() *http.ServeMux { + m := http.NewServeMux() + m.HandleFunc("GET /healthz", healthz) + return m +} + +func adminMux() *http.ServeMux { + m := baseMux() + m.HandleFunc("GET /admin/ping", func(w http.ResponseWriter, _ *http.Request) { _, _ = w.Write([]byte("pong\n")) }) + return m +} + +// Start binds all three addresses first, so a failure leaves nothing half-running. +func (s *Server) Start() error { + specs := []struct { + name, addr string + h http.Handler + }{ + {"public", s.cfg.PublicAddr, baseMux()}, + {"ingest", s.cfg.IngestAddr, baseMux()}, + {"admin", s.cfg.AdminAddr, adminMux()}, + } + for _, sp := range specs { + ln, err := net.Listen("tcp", sp.addr) + if err != nil { + s.closeListeners() + return fmt.Errorf("listen %s on %s: %w", sp.name, sp.addr, err) + } + srv := &http.Server{ + Handler: sp.h, + ReadHeaderTimeout: 10 * time.Second, + ReadTimeout: 30 * time.Second, + WriteTimeout: 30 * time.Second, + IdleTimeout: 60 * time.Second, + MaxHeaderBytes: 16 << 10, + } + s.ls = append(s.ls, &listener{name: sp.name, srv: srv, ln: ln}) + } + for _, l := range s.ls { + s.log.Info("listening", "listener", l.name, "addr", l.ln.Addr().String()) + go func(l *listener) { + if err := l.srv.Serve(l.ln); err != nil && !errors.Is(err, http.ErrServerClosed) { + s.log.Error("serve failed", "listener", l.name, "err", err) + s.errc <- fmt.Errorf("serve %s: %w", l.name, err) + } + }(l) + } + return nil +} + +func (s *Server) closeListeners() { + for _, l := range s.ls { + _ = l.ln.Close() + } + s.ls = nil +} + +// Addrs returns the bound addresses; call after Start. +func (s *Server) Addrs() Addrs { + var a Addrs + for _, l := range s.ls { + switch l.name { + case "public": + a.Public = l.ln.Addr().String() + case "ingest": + a.Ingest = l.ln.Addr().String() + case "admin": + a.Admin = l.ln.Addr().String() + } + } + return a +} + +// Err delivers the first unexpected listener failure; the caller should then Shutdown and exit nonzero. +func (s *Server) Err() <-chan error { return s.errc } + +// Shutdown stops accepting on all listeners at once, then waits for in-flight requests until ctx ends. +func (s *Server) Shutdown(ctx context.Context) error { + errs := make([]error, len(s.ls)) + var wg sync.WaitGroup + for i, l := range s.ls { + wg.Add(1) + go func() { + defer wg.Done() + if err := l.srv.Shutdown(ctx); err != nil { + errs[i] = fmt.Errorf("shutdown %s: %w", l.name, err) + } + }() + } + wg.Wait() + return errors.Join(errs...) +} diff --git a/internal/server/server_test.go b/internal/server/server_test.go new file mode 100644 index 0000000..0cdf73c --- /dev/null +++ b/internal/server/server_test.go @@ -0,0 +1,86 @@ +package server + +import ( + "context" + "io" + "log/slog" + "net/http" + "testing" + "time" + + "github.com/opensawce/kaasu/internal/config" +) + +func start(t *testing.T) *Server { + t.Helper() + cfg := config.Config{PublicAddr: "127.0.0.1:0", IngestAddr: "127.0.0.1:0", AdminAddr: "127.0.0.1:0"} + s := New(cfg, slog.New(slog.NewJSONHandler(io.Discard, nil))) + if err := s.Start(); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + _ = s.Shutdown(ctx) + }) + return s +} + +func get(t *testing.T, url string) (int, string) { + t.Helper() + resp, err := http.Get(url) + if err != nil { + t.Fatal(err) + } + defer resp.Body.Close() + b, _ := io.ReadAll(resp.Body) + return resp.StatusCode, string(b) +} + +func TestHealthzOnEveryListener(t *testing.T) { + s := start(t) + a := s.Addrs() + for name, addr := range map[string]string{"public": a.Public, "ingest": a.Ingest, "admin": a.Admin} { + code, body := get(t, "http://"+addr+"/healthz") + if code != 200 || body != `{"status":"ok"}`+"\n" { + t.Errorf("%s: code=%d body=%q", name, code, body) + } + } +} + +// Threat T6: admin routes must not exist on the public or ingest listeners. +func TestAdminRoutesOnlyOnAdminListener(t *testing.T) { + s := start(t) + a := s.Addrs() + if code, _ := get(t, "http://"+a.Admin+"/admin/ping"); code != 200 { + t.Errorf("admin listener should serve /admin/ping, got %d", code) + } + for name, addr := range map[string]string{"public": a.Public, "ingest": a.Ingest} { + if code, _ := get(t, "http://"+addr+"/admin/ping"); code != 404 { + t.Errorf("%s listener must not serve admin routes, got %d", name, code) + } + } +} + +func TestShutdownStopsListeners(t *testing.T) { + s := start(t) + addr := s.Addrs().Public + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + if err := s.Shutdown(ctx); err != nil { + t.Fatal(err) + } + if _, err := http.Get("http://" + addr + "/healthz"); err == nil { + t.Error("public listener still accepting after shutdown") + } +} + +func TestStartFailsWhenPortTaken(t *testing.T) { + s := start(t) + cfg := config.Config{PublicAddr: s.Addrs().Public, IngestAddr: "127.0.0.1:0", AdminAddr: "127.0.0.1:0"} + s2 := New(cfg, slog.New(slog.NewJSONHandler(io.Discard, nil))) + if err := s2.Start(); err == nil { + _ = s2.Shutdown(context.Background()) + t.Fatal("expected bind error") + } +} From 406100b57603291a68aca724fcea464d72d0a968 Mon Sep 17 00:00:00 2001 From: Manish <69393822+manish-9245@users.noreply.github.com> Date: Sat, 24 Oct 2026 12:05:42 +0530 Subject: [PATCH 02/54] chore(review): CodeRabbit rules for money path, security, customer pages and docs (#125) * chore(review): CodeRabbit rules for money path, security, customer pages and docs * fix: supported locale, no auto-pause, any base branch, parser fixtures stay in review --- .coderabbit.yaml | 117 +++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 117 insertions(+) create mode 100644 .coderabbit.yaml diff --git a/.coderabbit.yaml b/.coderabbit.yaml new file mode 100644 index 0000000..2f6ecc9 --- /dev/null +++ b/.coderabbit.yaml @@ -0,0 +1,117 @@ +# CodeRabbit review rules for Kaasu. Source of truth for review behaviour; AGENTS.md holds the project rules. +language: en-GB +early_access: false + +reviews: + profile: assertive + request_changes_workflow: false + high_level_summary: true + high_level_summary_placeholder: "@coderabbitai summary" + review_status: true + commit_status: true + collapse_walkthrough: false + poem: false + auto_review: + enabled: true + drafts: false + auto_incremental_review: true + auto_pause_after_reviewed_commits: 0 + base_branches: + - ".*" + + path_filters: + - "!**/*.golden" + - "!**/*.lock" + - "!**/package-lock.json" + - "!**/go.sum" + - "!**/*.png" + - "!**/*.webp" + + path_instructions: + - path: "internal/match/**" + instructions: >- + MONEY PATH. Only this package may confirm a payment. Check: amounts are int64 paise (never float), + an auto-confirm needs an exact amount AND an authentic source AND score >= 80 AND an order under the + account limit, late or closed orders go to review, and Process is idempotent. Flag any new code path + that can reach confirmed without these checks. Every behaviour change needs a test. + - path: "internal/states/**" + instructions: >- + MONEY PATH. The transition table is the only place order state changes. Check every new edge against the + table in the card, that each transition writes a ledger row in the same transaction, and that refunds are + the only release of a held amount. + - path: "internal/ledger/**" + instructions: >- + MONEY PATH. The ledger is append-only and hash-chained. Flag anything that updates or deletes rows, + changes the hash inputs, or stores raw alert text or full payer identifiers. + - path: "internal/ingest/**" + instructions: >- + SECURITY. Every request is signed over the body hash, timestamp and nonce; the window is 60 seconds; + replays get 409; every failure is the same 401 with no detail. Flag any path that accepts an unsigned + or stale request, and any use of == on secrets. + - path: "internal/parsers/**" + instructions: >- + MONEY PATH. Amounts are integer paise parsed from the digit string. Flag ParseFloat or any float + arithmetic on money, any parser that accepts a debit, and any fixture with real customer data. + - path: "internal/email/**" + instructions: >- + SECURITY. DKIM is verified on the raw bytes; the signing domain must be an allow-listed bank domain + (exact or subdomain); the From address must align. Flag any path that trusts a message without DKIM. + - path: "internal/secrets/**" + instructions: >- + SECURITY. Secrets are envelope-encrypted (XChaCha20-Poly1305); the master key never lives in the + database; API keys are stored as SHA-256 hashes and shown once. Flag plaintext storage or logging. + - path: "internal/adminauth/**" + instructions: >- + SECURITY. Sessions are stored as hashes, cookies are HttpOnly, SameSite=Strict and Secure outside local + http; TOTP codes are single-use; lockout and the same reply for unknown names must hold. Flag any change + that weakens these. + - path: "internal/web/**" + instructions: >- + CUSTOMER PAGES. No third-party scripts, no inline scripts, no 'failed' copy while an outcome is + unknown, and no Paisu on customer pages. Text comes from the i18n catalogue, never hard-coded. + - path: "i18n/**" + instructions: >- + Every key must exist in every language with the same placeholders. Machine drafts must stay listed in + i18n/REVIEW.md. Do not mark a translation reviewed. + - path: "**/*.md" + instructions: >- + Documentation. Flag em dashes (U+2014) and claims the code does not back up. Keep copy plain. + - path: "web/**" + instructions: >- + Website. No em dashes. No emoji used as icons. No tech-stack wording on the landing page; talk about impact. + - path: "**/*.go" + instructions: >- + Go: stdlib first. Wrap errors with context. No package-level mutable globals. Table-driven tests. Money is + int64 paise. Flag any network call in tests. + + tools: + golangci-lint: + enabled: false + gitleaks: + enabled: true + semgrep: + enabled: true + ast-grep: + enabled: false + + finishing_touches: + docstrings: + enabled: false + unit_tests: + enabled: false + +chat: + auto_reply: true + +knowledge_base: + opt_out: false + learnings: + scope: auto + code_guidelines: + enabled: true + filePatterns: + - "AGENTS.md" + - "CONTRIBUTING.md" + - "architecture.md" + - "design.md" + - "SECURITY.md" From 651d90ac0614d625cf8e32a4999882d408ba82a1 Mon Sep 17 00:00:00 2001 From: Manish <69393822+manish-9245@users.noreply.github.com> Date: Sat, 24 Oct 2026 12:11:44 +0530 Subject: [PATCH 03/54] feat(server): embedded SQLite with migrations (#88) * feat(server): embedded SQLite with migrations WAL, STRICT tables, foreign keys, single BEGIN IMMEDIATE writer with a separate read pool, embedded numbered migrations and kaasu migrate. * fix: read-only reader pool, rollback on panic, isolate reservation test * fix(deps): build with Go 1.27.3 to clear standard library vulnerabilities --- architecture.md | 14 ++ cmd/kaasu/main.go | 25 +++- cmd/kaasu/main_test.go | 20 +++ go.mod | 15 +- go.sum | 20 +++ internal/store/migrations/0001_init.sql | 119 ++++++++++++++++ internal/store/store.go | 124 +++++++++++++++++ internal/store/store_test.go | 178 ++++++++++++++++++++++++ 8 files changed, 513 insertions(+), 2 deletions(-) create mode 100644 go.sum create mode 100644 internal/store/migrations/0001_init.sql create mode 100644 internal/store/store.go create mode 100644 internal/store/store_test.go diff --git a/architecture.md b/architecture.md index dd297f8..d763b1f 100644 --- a/architecture.md +++ b/architecture.md @@ -22,6 +22,20 @@ Diagrams are drawn in Eraser. Source: `assets/diagrams/*.eraser` ยท Images: `ass ### Data model ![Data model](assets/diagrams/data-model.png) +SQLite in WAL mode, STRICT tables, foreign keys on, one writer (`BEGIN IMMEDIATE`). Migrations are numbered files in `internal/store/migrations/`; never edit an applied one. + +| Table | Holds | Guarded by | +|---|---|---| +| `receiving_accounts` | UPI IDs the shop receives on | unique VPA | +| `orders` | one row per order, state, exact paise | unique idempotency key, state check | +| `amount_reservations` | the unique amount an open order holds | primary key (account, amount) | +| `devices`, `nonces` | paired phones and replay protection | unique (device, nonce) | +| `evidence`, `alerts` | raw alert bodies and parsed facts | unique raw hash, unique UTR | +| `matches` | alert to order decisions | one match per alert | +| `refunds` | refunds recorded against orders | foreign key to orders | +| `ledger` | append-only hash chain | see ledger rules | +| `outbox` | webhook events waiting for delivery | unique event id | + ### Deployment | Option | Diagram | |---|---| diff --git a/cmd/kaasu/main.go b/cmd/kaasu/main.go index 5988a0a..fdb97d0 100644 --- a/cmd/kaasu/main.go +++ b/cmd/kaasu/main.go @@ -13,6 +13,7 @@ import ( "github.com/opensawce/kaasu/internal/config" "github.com/opensawce/kaasu/internal/server" + "github.com/opensawce/kaasu/internal/store" ) // version is set at build time with -ldflags "-X main.version=...". @@ -35,7 +36,9 @@ func run(args []string, getenv func(string) string, stdout, stderr io.Writer) in return 0 case "serve": return serve(args[1:], getenv, stderr) - case "migrate", "verify", "backup", "pair": + case "migrate": + return migrate(getenv, stdout, stderr) + case "verify", "backup", "pair": fmt.Fprintf(stderr, "kaasu %s: not implemented yet\n", args[0]) return 1 default: @@ -44,6 +47,26 @@ func run(args []string, getenv func(string) string, stdout, stderr io.Writer) in } } +func migrate(getenv func(string) string, stdout, stderr io.Writer) int { + cfg, err := config.Load(getenv, getenv("KAASU_CONFIG")) + if err != nil { + fmt.Fprintln(stderr, "config:", err) + return 1 + } + st, err := store.Open(cfg.DatabasePath) + if err != nil { + fmt.Fprintln(stderr, "open database:", err) + return 1 + } + defer st.Close() + if err := st.Migrate(context.Background()); err != nil { + fmt.Fprintln(stderr, "migrate:", err) + return 1 + } + fmt.Fprintln(stdout, "schema up to date") + return 0 +} + func serve(args []string, getenv func(string) string, stderr io.Writer) int { if len(args) > 0 { fmt.Fprintf(stderr, "serve: unexpected arguments %q (configure with KAASU_CONFIG or KAASU_* env)\n", args) diff --git a/cmd/kaasu/main_test.go b/cmd/kaasu/main_test.go index 0581a46..7b69cc7 100644 --- a/cmd/kaasu/main_test.go +++ b/cmd/kaasu/main_test.go @@ -2,6 +2,7 @@ package main import ( "bytes" + "path/filepath" "strings" "testing" ) @@ -36,3 +37,22 @@ func TestRun(t *testing.T) { }) } } + +func TestMigrateIsIdempotent(t *testing.T) { + db := filepath.Join(t.TempDir(), "k.db") + getenv := func(k string) string { + if k == "KAASU_DATABASE_PATH" { + return db + } + return "" + } + for i := 0; i < 2; i++ { + var out, errb bytes.Buffer + if code := run([]string{"migrate"}, getenv, &out, &errb); code != 0 { + t.Fatalf("run %d: code=%d stderr=%q", i, code, errb.String()) + } + if !strings.Contains(out.String(), "schema up to date") { + t.Errorf("run %d: stdout=%q", i, out.String()) + } + } +} diff --git a/go.mod b/go.mod index 76b25c6..cd4017b 100644 --- a/go.mod +++ b/go.mod @@ -1,3 +1,16 @@ module github.com/opensawce/kaasu -go 1.25 +go 1.27.3 + +require ( + github.com/dustin/go-humanize v1.1.0 // indirect + github.com/google/uuid v1.6.0 // indirect + github.com/mattn/go-isatty v0.0.24 // indirect + github.com/ncruces/go-strftime v1.1.0 // indirect + github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect + golang.org/x/sys v0.49.0 // indirect + modernc.org/libc v1.79.0 // indirect + modernc.org/mathutil v1.7.1 // indirect + modernc.org/memory v1.12.1 // indirect + modernc.org/sqlite v1.61.0 // indirect +) diff --git a/go.sum b/go.sum new file mode 100644 index 0000000..1e9e49e --- /dev/null +++ b/go.sum @@ -0,0 +1,20 @@ +github.com/dustin/go-humanize v1.1.0 h1:dbKTrvD0klcbBV/h4AWJdMuZogJACoMlvWIWZ5b2xWg= +github.com/dustin/go-humanize v1.1.0/go.mod h1:hc1CvRkJMsgxqjmjMQF3QNRAZBwY8AXBAzKYoSX9sFI= +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/mattn/go-isatty v0.0.24 h1:tGZZoVgT/KiqK1c8ocVLeDS8BSWMRd47J3Lbz7vsReI= +github.com/mattn/go-isatty v0.0.24/go.mod h1:nMCL3Zebbrt45jsMDgnfIwz6ydEQApk5oEI3HqDio6A= +github.com/ncruces/go-strftime v1.1.0 h1:vtPmPWMkb539RGV9WpWT9EwJGvrOCKvMNI4obWL47Hk= +github.com/ncruces/go-strftime v1.1.0/go.mod h1:yVaq7iwCN09vHRrKBrrflXT0M9tTMG2KmZ9Fj7xeG/g= +github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec h1:W09IVJc94icq4NjY3clb7Lk8O1qJ8BdBEF8z0ibU0rE= +github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo= +golang.org/x/sys v0.49.0 h1:XbzkgYJHdqh/8m2Uu0W/dQv8nktxx4BFHp1M0gROTXA= +golang.org/x/sys v0.49.0/go.mod h1:hNLxWAXmnKAxqDtdwIYC4bM9oQPEecfsnNMuSxOs3og= +modernc.org/libc v1.79.0 h1:9Dvvv++bAMSkSuBzuOFAatbTV5OIoq6Y6FohmD3e7q8= +modernc.org/libc v1.79.0/go.mod h1:uT3kR61NsbeKJSLqq83GJwSnZMVLlSjyAz4DfTaBy1M= +modernc.org/mathutil v1.7.1 h1:GCZVGXdaN8gTqB1Mf/usp1Y/hSqgI2vAGGP4jZMCxOU= +modernc.org/mathutil v1.7.1/go.mod h1:4p5IwJITfppl0G4sUEDtCr4DthTaT47/N3aT6MhfgJg= +modernc.org/memory v1.12.1 h1:nFMiWrpStgZczNl6XI9GnIk/rWhYIyHGUaR04pGbp9g= +modernc.org/memory v1.12.1/go.mod h1:/JP4VbVC+K5sU2wZi9bHoq2MAkCnrt2r98UGeSK7Mjw= +modernc.org/sqlite v1.61.0 h1:sQi63oLQ9uquAUA4CzOisH35Q3pEszkPBuHM0tFL8CY= +modernc.org/sqlite v1.61.0/go.mod h1:f4gr5tbpx2umklaT/QVtbi/50hNdM+4ng+2zMo8KXJ0= diff --git a/internal/store/migrations/0001_init.sql b/internal/store/migrations/0001_init.sql new file mode 100644 index 0000000..dc34004 --- /dev/null +++ b/internal/store/migrations/0001_init.sql @@ -0,0 +1,119 @@ +-- Money is always integer paise. Times are RFC 3339 UTC text. Never edit this file; add a new migration. + +CREATE TABLE receiving_accounts ( + id TEXT PRIMARY KEY, + vpa TEXT NOT NULL UNIQUE, + bank TEXT NOT NULL, + display_name TEXT NOT NULL, + auto_confirm_limit_paise INTEGER NOT NULL DEFAULT 0 CHECK (auto_confirm_limit_paise >= 0), + active INTEGER NOT NULL DEFAULT 1 CHECK (active IN (0, 1)), + created_at TEXT NOT NULL +) STRICT; + +CREATE TABLE orders ( + id TEXT PRIMARY KEY, + account_id TEXT NOT NULL REFERENCES receiving_accounts(id), + base_amount_paise INTEGER NOT NULL CHECK (base_amount_paise > 0), + amount_paise INTEGER NOT NULL CHECK (amount_paise > 0), + currency TEXT NOT NULL DEFAULT 'INR' CHECK (currency = 'INR'), + state TEXT NOT NULL CHECK (state IN ('created','queued','awaiting','matched','confirmed','review','rejected','cancelled','expired','late_paid','refunded','disputed')), + reference TEXT, + note TEXT, + idempotency_key TEXT UNIQUE, + request_hash TEXT, + expires_at TEXT NOT NULL, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL, + confirmed_at TEXT +) STRICT; +CREATE INDEX orders_account_state ON orders(account_id, state); + +CREATE TABLE amount_reservations ( + account_id TEXT NOT NULL REFERENCES receiving_accounts(id), + amount_paise INTEGER NOT NULL CHECK (amount_paise > 0), + order_id TEXT NOT NULL UNIQUE REFERENCES orders(id), + reserved_until TEXT NOT NULL, + PRIMARY KEY (account_id, amount_paise) +) STRICT; + +CREATE TABLE devices ( + id TEXT PRIMARY KEY, + label TEXT NOT NULL, + kind TEXT NOT NULL CHECK (kind IN ('android','apple','windows','email')), + public_key BLOB NOT NULL, + paired_at TEXT NOT NULL, + last_seen_at TEXT, + revoked_at TEXT +) STRICT; + +CREATE TABLE nonces ( + device_id TEXT NOT NULL REFERENCES devices(id), + nonce TEXT NOT NULL, + seen_at TEXT NOT NULL, + PRIMARY KEY (device_id, nonce) +) STRICT; + +CREATE TABLE evidence ( + id TEXT PRIMARY KEY, + source TEXT NOT NULL CHECK (source IN ('device','email','manual')), + device_id TEXT REFERENCES devices(id), + received_at TEXT NOT NULL, + raw_hash TEXT NOT NULL UNIQUE, + body BLOB NOT NULL +) STRICT; + +CREATE TABLE alerts ( + id TEXT PRIMARY KEY, + evidence_id TEXT NOT NULL REFERENCES evidence(id), + bank TEXT, + amount_paise INTEGER NOT NULL CHECK (amount_paise > 0), + utr TEXT UNIQUE, + payer_masked TEXT, + account_hint TEXT, + occurred_at TEXT NOT NULL, + parsed_at TEXT NOT NULL, + parser TEXT +) STRICT; + +CREATE TABLE matches ( + id TEXT PRIMARY KEY, + alert_id TEXT NOT NULL UNIQUE REFERENCES alerts(id), + order_id TEXT NOT NULL REFERENCES orders(id), + score INTEGER NOT NULL CHECK (score BETWEEN 0 AND 100), + decision TEXT NOT NULL CHECK (decision IN ('auto','review','rejected')), + decided_by TEXT, + created_at TEXT NOT NULL +) STRICT; + +CREATE TABLE refunds ( + id TEXT PRIMARY KEY, + order_id TEXT NOT NULL REFERENCES orders(id), + amount_paise INTEGER NOT NULL CHECK (amount_paise > 0), + utr TEXT, + note TEXT, + created_at TEXT NOT NULL +) STRICT; + +CREATE TABLE ledger ( + seq INTEGER PRIMARY KEY AUTOINCREMENT, + ts TEXT NOT NULL, + kind TEXT NOT NULL, + order_id TEXT, + actor TEXT NOT NULL, + payload TEXT NOT NULL, + prev_hash TEXT NOT NULL, + hash TEXT NOT NULL UNIQUE +) STRICT; + +CREATE TABLE outbox ( + id TEXT PRIMARY KEY, + event_id TEXT NOT NULL UNIQUE, + kind TEXT NOT NULL, + payload TEXT NOT NULL, + status TEXT NOT NULL DEFAULT 'pending' CHECK (status IN ('pending','delivered','dead')), + attempts INTEGER NOT NULL DEFAULT 0, + next_attempt_at TEXT NOT NULL, + created_at TEXT NOT NULL, + delivered_at TEXT +) STRICT; +CREATE INDEX outbox_due ON outbox(status, next_attempt_at); diff --git a/internal/store/store.go b/internal/store/store.go new file mode 100644 index 0000000..45e6228 --- /dev/null +++ b/internal/store/store.go @@ -0,0 +1,124 @@ +// Package store owns the SQLite database: connection setup, migrations and the single-writer transaction helper. +package store + +import ( + "context" + "database/sql" + "embed" + "errors" + "fmt" + "net/url" + "sort" + "strconv" + "strings" + "time" + + _ "modernc.org/sqlite" // pure Go driver, keeps the binary static +) + +//go:embed migrations/*.sql +var migrationFiles embed.FS + +// Store is one database file. Writes go through one connection (BEGIN IMMEDIATE); reads use their own pool. +type Store struct { + db *sql.DB // single writer + ro *sql.DB // concurrent readers +} + +func dsn(path, extra string) string { + q := url.Values{} + q.Add("_pragma", "journal_mode(WAL)") + q.Add("_pragma", "foreign_keys(1)") + q.Add("_pragma", "busy_timeout(5000)") + q.Add("_pragma", "synchronous(NORMAL)") + if extra != "" { + k, v, _ := strings.Cut(extra, "=") + q.Add(k, v) + } + return "file:" + path + "?" + q.Encode() +} + +// Open opens (and creates) the database at path. +func Open(path string) (*Store, error) { + w, err := sql.Open("sqlite", dsn(path, "_txlock=immediate")) + if err != nil { + return nil, fmt.Errorf("open writer: %w", err) + } + w.SetMaxOpenConns(1) + if err := w.Ping(); err != nil { + _ = w.Close() + return nil, fmt.Errorf("open %s: %w", path, err) + } + r, err := sql.Open("sqlite", dsn(path, "_pragma=query_only(1)")) + if err != nil { + _ = w.Close() + return nil, fmt.Errorf("open reader: %w", err) + } + r.SetMaxOpenConns(8) + return &Store{db: w, ro: r}, nil +} + +func (s *Store) Close() error { return errors.Join(s.ro.Close(), s.db.Close()) } + +// Read returns the read pool. Never write through it. +func (s *Store) Read() *sql.DB { return s.ro } + +// Tx runs fn in a BEGIN IMMEDIATE transaction on the single writer; it commits unless fn fails. +func (s *Store) Tx(ctx context.Context, fn func(*sql.Tx) error) error { + tx, err := s.db.BeginTx(ctx, nil) + if err != nil { + return fmt.Errorf("begin: %w", err) + } + defer func() { _ = tx.Rollback() }() // no-op after Commit; releases the writer if fn panics + if err := fn(tx); err != nil { + return err + } + if err := tx.Commit(); err != nil { + return fmt.Errorf("commit: %w", err) + } + return nil +} + +// Migrate applies every embedded migration that has not run yet. Safe to call repeatedly. +func (s *Store) Migrate(ctx context.Context) error { + if _, err := s.db.ExecContext(ctx, `CREATE TABLE IF NOT EXISTS schema_migrations (version INTEGER PRIMARY KEY, name TEXT NOT NULL, applied_at TEXT NOT NULL) STRICT`); err != nil { + return fmt.Errorf("create schema_migrations: %w", err) + } + entries, err := migrationFiles.ReadDir("migrations") + if err != nil { + return err + } + names := make([]string, 0, len(entries)) + for _, e := range entries { + names = append(names, e.Name()) + } + sort.Strings(names) + for _, name := range names { + version, err := strconv.Atoi(strings.SplitN(name, "_", 2)[0]) + if err != nil { + return fmt.Errorf("migration %s: name must start with a number: %w", name, err) + } + body, err := migrationFiles.ReadFile("migrations/" + name) + if err != nil { + return err + } + err = s.Tx(ctx, func(tx *sql.Tx) error { + var done int + if err := tx.QueryRowContext(ctx, `SELECT count(*) FROM schema_migrations WHERE version = ?`, version).Scan(&done); err != nil { + return err + } + if done > 0 { + return nil + } + if _, err := tx.ExecContext(ctx, string(body)); err != nil { + return fmt.Errorf("apply: %w", err) + } + _, err := tx.ExecContext(ctx, `INSERT INTO schema_migrations(version,name,applied_at) VALUES(?,?,?)`, version, name, time.Now().UTC().Format(time.RFC3339)) + return err + }) + if err != nil { + return fmt.Errorf("migration %s: %w", name, err) + } + } + return nil +} diff --git a/internal/store/store_test.go b/internal/store/store_test.go new file mode 100644 index 0000000..a2b054f --- /dev/null +++ b/internal/store/store_test.go @@ -0,0 +1,178 @@ +package store + +import ( + "context" + "database/sql" + "fmt" + "path/filepath" + "strings" + "sync" + "testing" +) + +func open(t *testing.T) *Store { + t.Helper() + s, err := Open(filepath.Join(t.TempDir(), "kaasu.db")) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = s.Close() }) + if err := s.Migrate(context.Background()); err != nil { + t.Fatal(err) + } + return s +} + +func pragma(t *testing.T, s *Store, name string) string { + t.Helper() + var v string + if err := s.db.QueryRow("PRAGMA " + name).Scan(&v); err != nil { + t.Fatal(err) + } + return v +} + +func TestOpenSetsWALAndForeignKeys(t *testing.T) { + s := open(t) + if got := pragma(t, s, "journal_mode"); got != "wal" { + t.Errorf("journal_mode=%s", got) + } + if got := pragma(t, s, "foreign_keys"); got != "1" { + t.Errorf("foreign_keys=%s", got) + } +} + +func TestMigrateCreatesStrictTables(t *testing.T) { + s := open(t) + want := []string{"receiving_accounts", "orders", "amount_reservations", "devices", "nonces", "evidence", "alerts", "matches", "refunds", "ledger", "outbox"} + for _, name := range want { + var strict int + err := s.db.QueryRow(`SELECT strict FROM pragma_table_list WHERE name = ?`, name).Scan(&strict) + if err != nil { + t.Errorf("table %s missing: %v", name, err) + continue + } + if strict != 1 { + t.Errorf("table %s is not STRICT", name) + } + } +} + +func TestMigrateIsIdempotent(t *testing.T) { + s := open(t) + var before int + _ = s.db.QueryRow(`SELECT count(*) FROM schema_migrations`).Scan(&before) + if err := s.Migrate(context.Background()); err != nil { + t.Fatal(err) + } + var after int + _ = s.db.QueryRow(`SELECT count(*) FROM schema_migrations`).Scan(&after) + if before == 0 || before != after { + t.Errorf("migrations before=%d after=%d", before, after) + } +} + +func seed(t *testing.T, s *Store) { + t.Helper() + stmts := []string{ + `INSERT INTO receiving_accounts(id,vpa,bank,display_name,created_at) VALUES('acc1','shop@okbank','Bank','Shop','2026-01-01T00:00:00Z')`, + `INSERT INTO orders(id,account_id,base_amount_paise,amount_paise,state,idempotency_key,request_hash,expires_at,created_at,updated_at) + VALUES('o1','acc1',49900,49907,'awaiting','k1','h1','2026-01-01T00:15:00Z','2026-01-01T00:00:00Z','2026-01-01T00:00:00Z')`, + `INSERT INTO orders(id,account_id,base_amount_paise,amount_paise,state,idempotency_key,request_hash,expires_at,created_at,updated_at) + VALUES('o3','acc1',49900,49908,'awaiting','k3','h3','2026-01-01T00:15:00Z','2026-01-01T00:00:00Z','2026-01-01T00:00:00Z')`, + `INSERT INTO devices(id,label,kind,public_key,paired_at) VALUES('d1','Phone','android',x'00','2026-01-01T00:00:00Z')`, + `INSERT INTO evidence(id,source,device_id,received_at,raw_hash,body) VALUES('e1','device','d1','2026-01-01T00:01:00Z','rh1',x'00')`, + `INSERT INTO alerts(id,evidence_id,amount_paise,utr,occurred_at,parsed_at) VALUES('a1','e1',49907,'UTR1','2026-01-01T00:01:00Z','2026-01-01T00:01:01Z')`, + } + for _, q := range stmts { + if _, err := s.db.Exec(q); err != nil { + t.Fatalf("seed %q: %v", q, err) + } + } +} + +func TestConstraints(t *testing.T) { + s := open(t) + seed(t, s) + tests := []struct{ name, sql string }{ + {"reservation unique per account and amount", `INSERT INTO amount_reservations(account_id,amount_paise,order_id,reserved_until) VALUES('acc1',49907,'o3','2026-01-01T01:00:00Z')`}, + {"UTR unique", `INSERT INTO alerts(id,evidence_id,amount_paise,utr,occurred_at,parsed_at) VALUES('a2','e1',100,'UTR1','2026-01-01T00:02:00Z','2026-01-01T00:02:01Z')`}, + {"one match per alert", `INSERT INTO matches(id,alert_id,order_id,score,decision,created_at) VALUES('m2','a1','o1',90,'auto','2026-01-01T00:03:00Z')`}, + {"nonce unique per device", `INSERT INTO nonces(device_id,nonce,seen_at) VALUES('d1','n1','2026-01-01T00:00:00Z')`}, + {"idempotency key unique", `INSERT INTO orders(id,account_id,base_amount_paise,amount_paise,state,idempotency_key,request_hash,expires_at,created_at,updated_at) + VALUES('o2','acc1',100,100,'awaiting','k1','h2','2026-01-01T00:15:00Z','2026-01-01T00:00:00Z','2026-01-01T00:00:00Z')`}, + {"foreign key enforced", `INSERT INTO refunds(id,order_id,amount_paise,created_at) VALUES('r1','nope',1,'2026-01-01T00:00:00Z')`}, + {"state check", `UPDATE orders SET state='bogus' WHERE id='o1'`}, + {"amounts positive", `UPDATE orders SET amount_paise=0 WHERE id='o1'`}, + } + // First insert the rows that the duplicates collide with. + for _, q := range []string{ + `INSERT INTO amount_reservations(account_id,amount_paise,order_id,reserved_until) VALUES('acc1',49907,'o1','2026-01-01T01:00:00Z')`, + `INSERT INTO matches(id,alert_id,order_id,score,decision,created_at) VALUES('m1','a1','o1',90,'auto','2026-01-01T00:03:00Z')`, + `INSERT INTO nonces(device_id,nonce,seen_at) VALUES('d1','n1','2026-01-01T00:00:00Z')`, + } { + if _, err := s.db.Exec(q); err != nil { + t.Fatalf("setup %q: %v", q, err) + } + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if _, err := s.db.Exec(tt.sql); err == nil { + t.Error("expected constraint error") + } + }) + } +} + +func TestTxSerialisesWriters(t *testing.T) { + s := open(t) + if _, err := s.db.Exec(`CREATE TABLE counter(n INTEGER NOT NULL) STRICT; INSERT INTO counter VALUES(0)`); err != nil { + t.Fatal(err) + } + var wg sync.WaitGroup + errs := make(chan error, 40) + for i := 0; i < 40; i++ { + wg.Add(1) + go func() { + defer wg.Done() + errs <- s.Tx(context.Background(), func(tx *sql.Tx) error { + var n int + if err := tx.QueryRow(`SELECT n FROM counter`).Scan(&n); err != nil { + return err + } + _, err := tx.Exec(`UPDATE counter SET n = ?`, n+1) + return err + }) + }() + } + wg.Wait() + close(errs) + for err := range errs { + if err != nil { + t.Fatal(err) + } + } + var n int + _ = s.db.QueryRow(`SELECT n FROM counter`).Scan(&n) + if n != 40 { + t.Errorf("lost updates: n=%d", n) + } +} + +func TestTxRollsBackOnError(t *testing.T) { + s := open(t) + err := s.Tx(context.Background(), func(tx *sql.Tx) error { + if _, err := tx.Exec(`INSERT INTO receiving_accounts(id,vpa,bank,display_name,created_at) VALUES('x','x@y','b','n','t')`); err != nil { + return err + } + return fmt.Errorf("boom") + }) + if err == nil || !strings.Contains(err.Error(), "boom") { + t.Fatalf("err=%v", err) + } + var n int + _ = s.db.QueryRow(`SELECT count(*) FROM receiving_accounts`).Scan(&n) + if n != 0 { + t.Error("insert survived rollback") + } +} From b69d42276bad35b5df4285103b8e3848e1201196 Mon Sep 17 00:00:00 2001 From: Manish <69393822+manish-9245@users.noreply.github.com> Date: Sat, 24 Oct 2026 12:13:13 +0530 Subject: [PATCH 04/54] ci: lint, race tests, coverage gate, fuzz and cross-builds (#89) * ci: lint, test with race and coverage gate, fuzz, and cross-build jobs * fix: fuzz script stops if package discovery fails (coverage YAML comment is stale: go.yml has no such line) --- .github/workflows/go.yml | 29 +++++++++++++++++++++++++++++ .gitignore | 1 + Makefile | 8 +++++++- scripts/coverage-gate.sh | 16 ++++++++++++++++ scripts/fuzz.sh | 15 +++++++++++++++ 5 files changed, 68 insertions(+), 1 deletion(-) create mode 100755 scripts/coverage-gate.sh create mode 100755 scripts/fuzz.sh diff --git a/.github/workflows/go.yml b/.github/workflows/go.yml index f731e23..a46d9a2 100644 --- a/.github/workflows/go.yml +++ b/.github/workflows/go.yml @@ -5,7 +5,36 @@ on: branches: [main] permissions: contents: read +concurrency: + group: go-${{ github.ref }} + cancel-in-progress: true jobs: + lint: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-go@v5 + with: { go-version-file: go.mod } + - name: gofmt + run: test -z "$(gofmt -l .)" || (gofmt -l . && exit 1) + - run: go vet ./... + - run: go run honnef.co/go/tools/cmd/staticcheck@2026.2.2 ./... + test: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-go@v5 + with: { go-version-file: go.mod } + - run: make test-cover + - uses: actions/upload-artifact@v4 + with: { name: coverage, path: coverage.out, retention-days: 7 } + fuzz: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-go@v5 + with: { go-version-file: go.mod } + - run: scripts/fuzz.sh 30s build: runs-on: ubuntu-latest strategy: diff --git a/.gitignore b/.gitignore index 2bf38d4..cc034ee 100644 --- a/.gitignore +++ b/.gitignore @@ -3,3 +3,4 @@ bin/ *.db-wal *.db-shm .agents/plans/ +coverage.out diff --git a/Makefile b/Makefile index 90a5692..faf2c6d 100644 --- a/Makefile +++ b/Makefile @@ -1,10 +1,16 @@ -.PHONY: dev test lint build +.PHONY: dev test test-cover lint build fuzz dev: go run ./cmd/kaasu serve test: go test -race ./... +test-cover: + go test -race -coverprofile=coverage.out ./... + scripts/coverage-gate.sh lint: @test -z "$$(gofmt -l .)" || (gofmt -l . && exit 1) go vet ./... + go run honnef.co/go/tools/cmd/staticcheck@2026.2.2 ./... +fuzz: + scripts/fuzz.sh 20s build: CGO_ENABLED=0 go build -trimpath -ldflags "-s -w" -o bin/kaasu ./cmd/kaasu diff --git a/scripts/coverage-gate.sh b/scripts/coverage-gate.sh new file mode 100755 index 0000000..d217c43 --- /dev/null +++ b/scripts/coverage-gate.sh @@ -0,0 +1,16 @@ +#!/bin/sh +# Fails if a money-path package that exists has less than MIN% statement coverage (default 85). +set -eu +min=${MIN:-85} +fail=0 +for p in orders pool states ledger match; do + dir="internal/$p" + [ -d "$dir" ] || continue + out=$(go test -cover "./$dir" 2>&1 | tail -1) + pct=$(echo "$out" | sed -n 's/.*coverage: \([0-9.]*\)%.*/\1/p') + [ -n "$pct" ] || { echo "$p: no coverage figure ($out)"; fail=1; continue; } + ok=$(awk -v a="$pct" -v b="$min" 'BEGIN{print (a>=b)?"ok":"LOW"}') + echo "$p: $pct% ($ok, need $min%)" + [ "$ok" = ok ] || fail=1 +done +exit $fail diff --git a/scripts/fuzz.sh b/scripts/fuzz.sh new file mode 100755 index 0000000..c893f66 --- /dev/null +++ b/scripts/fuzz.sh @@ -0,0 +1,15 @@ +#!/bin/sh +# Runs every Go fuzz target for the given time (default 20s each). Exits non-zero if any finds a crash. +set -eu +t=${1:-20s} +found=0 +pkgs=$(go list ./...) +for pkg in $pkgs; do + dir=$(go list -f '{{.Dir}}' "$pkg") + for f in $(grep -hoE '^func (Fuzz[A-Za-z0-9_]+)' "$dir"/*_test.go 2>/dev/null | awk '{print $2}'); do + found=1 + echo "fuzz $pkg $f for $t" + go test "$pkg" -run '^$' -fuzz "^${f}\$" -fuzztime "$t" + done +done +[ "$found" = 1 ] || echo "no fuzz targets yet" From 38a03880173791875e34fccd2392ddb4a3fde018 Mon Sep 17 00:00:00 2001 From: Manish <69393822+manish-9245@users.noreply.github.com> Date: Sat, 24 Oct 2026 12:15:15 +0530 Subject: [PATCH 05/54] test: shared test helpers, injectable clock and ids (#90) * test: clock, id and test helpers with golden files and migrated store * fix: valid YAML for coverage step, drop package-level base32 encoder --- .github/workflows/go.yml | 2 + internal/clock/clock.go | 41 ++++++++++++ internal/clock/clock_test.go | 47 ++++++++++++++ internal/id/id.go | 17 +++++ internal/id/id_test.go | 21 ++++++ internal/testutil/testdata/greeting.golden | 1 + internal/testutil/testutil.go | 74 ++++++++++++++++++++++ internal/testutil/testutil_test.go | 43 +++++++++++++ 8 files changed, 246 insertions(+) create mode 100644 internal/clock/clock.go create mode 100644 internal/clock/clock_test.go create mode 100644 internal/id/id.go create mode 100644 internal/id/id_test.go create mode 100644 internal/testutil/testdata/greeting.golden create mode 100644 internal/testutil/testutil.go create mode 100644 internal/testutil/testutil_test.go diff --git a/.github/workflows/go.yml b/.github/workflows/go.yml index a46d9a2..ac994b5 100644 --- a/.github/workflows/go.yml +++ b/.github/workflows/go.yml @@ -26,6 +26,8 @@ jobs: - uses: actions/setup-go@v5 with: { go-version-file: go.mod } - run: make test-cover + - run: | + go tool cover -func=coverage.out | tail -1 | sed 's/^/Total coverage: /' >> "$GITHUB_STEP_SUMMARY" - uses: actions/upload-artifact@v4 with: { name: coverage, path: coverage.out, retention-days: 7 } fuzz: diff --git a/internal/clock/clock.go b/internal/clock/clock.go new file mode 100644 index 0000000..a126409 --- /dev/null +++ b/internal/clock/clock.go @@ -0,0 +1,41 @@ +// Package clock lets code ask for the time without calling time.Now, so tests and dev mode can control it. +package clock + +import ( + "sync" + "time" +) + +// Clock is the only way production code reads the time. +type Clock interface{ Now() time.Time } + +// Real is wall-clock time in UTC. +type Real struct{} + +func (Real) Now() time.Time { return time.Now().UTC() } + +// Fake is a clock that only moves when told to. +type Fake struct { + mu sync.Mutex + t time.Time +} + +func NewFake(start time.Time) *Fake { return &Fake{t: start.UTC()} } + +func (f *Fake) Now() time.Time { + f.mu.Lock() + defer f.mu.Unlock() + return f.t +} + +func (f *Fake) Advance(d time.Duration) { + f.mu.Lock() + f.t = f.t.Add(d) + f.mu.Unlock() +} + +func (f *Fake) Set(t time.Time) { + f.mu.Lock() + f.t = t.UTC() + f.mu.Unlock() +} diff --git a/internal/clock/clock_test.go b/internal/clock/clock_test.go new file mode 100644 index 0000000..9d2aa3d --- /dev/null +++ b/internal/clock/clock_test.go @@ -0,0 +1,47 @@ +package clock + +import ( + "testing" + "time" +) + +func TestFakeAdvanceAndSet(t *testing.T) { + start := time.Date(2026, 1, 1, 10, 0, 0, 0, time.UTC) + f := NewFake(start) + if !f.Now().Equal(start) { + t.Fatalf("now=%v", f.Now()) + } + f.Advance(15 * time.Minute) + if want := start.Add(15 * time.Minute); !f.Now().Equal(want) { + t.Errorf("after advance now=%v want %v", f.Now(), want) + } + later := start.Add(24 * time.Hour) + f.Set(later) + if !f.Now().Equal(later) { + t.Errorf("after set now=%v", f.Now()) + } +} + +func TestRealIsUTCAndMoves(t *testing.T) { + a := Real{}.Now() + if a.Location() != time.UTC { + t.Errorf("location=%v", a.Location()) + } + if time.Since(a) > time.Second { + t.Error("real clock far from time.Now") + } +} + +func TestFakeConcurrentAdvance(t *testing.T) { + f := NewFake(time.Unix(0, 0).UTC()) + done := make(chan struct{}) + for i := 0; i < 50; i++ { + go func() { f.Advance(time.Second); done <- struct{}{} }() + } + for i := 0; i < 50; i++ { + <-done + } + if got := f.Now().Unix(); got != 50 { + t.Errorf("lost advances: %d", got) + } +} diff --git a/internal/id/id.go b/internal/id/id.go new file mode 100644 index 0000000..9877331 --- /dev/null +++ b/internal/id/id.go @@ -0,0 +1,17 @@ +// Package id makes short, prefixed, random identifiers such as ord_x7k2m9q4w8r5t1y3u6p0. +package id + +import ( + "crypto/rand" + "encoding/base32" + "strings" +) + +// New returns prefix + "_" + 20 random base32 characters (100 bits). +func New(prefix string) string { + b := make([]byte, 13) + if _, err := rand.Read(b); err != nil { + panic("id: crypto/rand failed: " + err.Error()) // no safe way to continue + } + return prefix + "_" + strings.ToLower(base32.HexEncoding.WithPadding(base32.NoPadding).EncodeToString(b))[:20] +} diff --git a/internal/id/id_test.go b/internal/id/id_test.go new file mode 100644 index 0000000..affb2c9 --- /dev/null +++ b/internal/id/id_test.go @@ -0,0 +1,21 @@ +package id + +import ( + "regexp" + "testing" +) + +func TestNewFormatAndUniqueness(t *testing.T) { + re := regexp.MustCompile(`^ord_[0-9a-z]{20}$`) + seen := map[string]bool{} + for i := 0; i < 10000; i++ { + v := New("ord") + if !re.MatchString(v) { + t.Fatalf("bad id %q", v) + } + if seen[v] { + t.Fatalf("duplicate id %q", v) + } + seen[v] = true + } +} diff --git a/internal/testutil/testdata/greeting.golden b/internal/testutil/testdata/greeting.golden new file mode 100644 index 0000000..ce01362 --- /dev/null +++ b/internal/testutil/testdata/greeting.golden @@ -0,0 +1 @@ +hello diff --git a/internal/testutil/testutil.go b/internal/testutil/testutil.go new file mode 100644 index 0000000..e33b685 --- /dev/null +++ b/internal/testutil/testutil.go @@ -0,0 +1,74 @@ +// Package testutil holds helpers shared by tests: deterministic keys and ids, a migrated store, and golden files. +package testutil + +import ( + "context" + "crypto/ed25519" + "flag" + "fmt" + "os" + "path/filepath" + "sync" + "testing" + + "github.com/opensawce/kaasu/internal/store" +) + +var update = flag.Bool("update", false, "rewrite golden files") + +// Key returns a deterministic Ed25519 key for a small seed. Test use only. +func Key(seed byte) ed25519.PrivateKey { + s := make([]byte, ed25519.SeedSize) + for i := range s { + s[i] = seed + } + return ed25519.NewKeyFromSeed(s) +} + +// Seq hands out predictable ids: prefix_000001, prefix_000002, ... +type Seq struct { + mu sync.Mutex + n int +} + +func (s *Seq) New(prefix string) string { + s.mu.Lock() + defer s.mu.Unlock() + s.n++ + return fmt.Sprintf("%s_%06d", prefix, s.n) +} + +// OpenStore returns a migrated SQLite store in a temp dir, closed when the test ends. +func OpenStore(t testing.TB) *store.Store { + t.Helper() + s, err := store.Open(filepath.Join(t.TempDir(), "kaasu.db")) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = s.Close() }) + if err := s.Migrate(context.Background()); err != nil { + t.Fatal(err) + } + return s +} + +// Golden compares got with testdata/.golden; run with -update to rewrite it. +func Golden(t testing.TB, name string, got []byte) { + t.Helper() + path := filepath.Join("testdata", name+".golden") + if *update { + if err := os.MkdirAll("testdata", 0o755); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(path, got, 0o644); err != nil { + t.Fatal(err) + } + } + want, err := os.ReadFile(path) + if err != nil { + t.Fatalf("missing golden file %s (run with -update): %v", path, err) + } + if string(want) != string(got) { + t.Errorf("golden mismatch for %s\n--- want\n%s\n--- got\n%s", name, want, got) + } +} diff --git a/internal/testutil/testutil_test.go b/internal/testutil/testutil_test.go new file mode 100644 index 0000000..c12b328 --- /dev/null +++ b/internal/testutil/testutil_test.go @@ -0,0 +1,43 @@ +package testutil + +import ( + "context" + "crypto/ed25519" + "testing" +) + +func TestKeyIsDeterministic(t *testing.T) { + a, b := Key(1), Key(1) + if !a.Equal(b) { + t.Error("same seed must give the same key") + } + if a.Equal(Key(2)) { + t.Error("different seeds must differ") + } + msg := []byte("hello") + if !ed25519.Verify(a.Public().(ed25519.PublicKey), msg, ed25519.Sign(a, msg)) { + t.Error("key does not sign") + } +} + +func TestSeqIDs(t *testing.T) { + s := &Seq{} + if got := s.New("ord"); got != "ord_000001" { + t.Errorf("got %q", got) + } + if got := s.New("evt"); got != "evt_000002" { + t.Errorf("got %q", got) + } +} + +func TestOpenStoreIsMigrated(t *testing.T) { + s := OpenStore(t) + var n int + if err := s.Read().QueryRowContext(context.Background(), `SELECT count(*) FROM schema_migrations`).Scan(&n); err != nil || n == 0 { + t.Fatalf("n=%d err=%v", n, err) + } +} + +func TestGoldenRoundTrip(t *testing.T) { + Golden(t, "greeting", []byte("hello\n")) +} From 11cbc5ea08d16abb7fc763fc2bdf5a67c17f8044 Mon Sep 17 00:00:00 2001 From: Manish <69393822+manish-9245@users.noreply.github.com> Date: Sat, 24 Oct 2026 12:17:10 +0530 Subject: [PATCH 06/54] feat(security): envelope encryption for secrets and API keys (#91) * feat(security): envelope-encrypted secrets, API keys and master key rotation * fix: check rows.Err in RotateAll --- cmd/kaasu/main.go | 49 ++++- cmd/kaasu/main_test.go | 32 ++++ go.mod | 1 + go.sum | 2 + internal/secrets/crypto.go | 175 ++++++++++++++++++ internal/secrets/key.go | 45 +++++ internal/secrets/secrets_test.go | 203 +++++++++++++++++++++ internal/secrets/vault.go | 175 ++++++++++++++++++ internal/store/migrations/0002_secrets.sql | 14 ++ 9 files changed, 695 insertions(+), 1 deletion(-) create mode 100644 internal/secrets/crypto.go create mode 100644 internal/secrets/key.go create mode 100644 internal/secrets/secrets_test.go create mode 100644 internal/secrets/vault.go create mode 100644 internal/store/migrations/0002_secrets.sql diff --git a/cmd/kaasu/main.go b/cmd/kaasu/main.go index fdb97d0..9017b62 100644 --- a/cmd/kaasu/main.go +++ b/cmd/kaasu/main.go @@ -3,6 +3,8 @@ package main import ( "context" + "crypto/rand" + "encoding/base64" "fmt" "io" "log/slog" @@ -11,7 +13,9 @@ import ( "syscall" "time" + "github.com/opensawce/kaasu/internal/clock" "github.com/opensawce/kaasu/internal/config" + "github.com/opensawce/kaasu/internal/secrets" "github.com/opensawce/kaasu/internal/server" "github.com/opensawce/kaasu/internal/store" ) @@ -19,7 +23,7 @@ import ( // version is set at build time with -ldflags "-X main.version=...". var version = "dev" -const usage = "usage: kaasu [flags]" +const usage = "usage: kaasu [flags]" func main() { os.Exit(run(os.Args[1:], os.Getenv, os.Stdout, os.Stderr)) @@ -38,6 +42,16 @@ func run(args []string, getenv func(string) string, stdout, stderr io.Writer) in return serve(args[1:], getenv, stderr) case "migrate": return migrate(getenv, stdout, stderr) + case "keygen": + k := make([]byte, 32) + if _, err := rand.Read(k); err != nil { + fmt.Fprintln(stderr, "keygen:", err) + return 1 + } + fmt.Fprintln(stdout, base64.StdEncoding.EncodeToString(k)) + return 0 + case "rotate-key": + return rotateKey(getenv, stdout, stderr) case "verify", "backup", "pair": fmt.Fprintf(stderr, "kaasu %s: not implemented yet\n", args[0]) return 1 @@ -67,6 +81,39 @@ func migrate(getenv func(string) string, stdout, stderr io.Writer) int { return 0 } +// rotateKey re-wraps every stored secret under KAASU_NEW_MASTER_KEY. After it succeeds, +// switch KAASU_MASTER_KEY to the new key; nothing changes if any secret fails to rewrap. +func rotateKey(getenv func(string) string, stdout, stderr io.Writer) int { + cfg, err := config.Load(getenv, getenv("KAASU_CONFIG")) + if err != nil { + fmt.Fprintln(stderr, "config:", err) + return 1 + } + cur, _, err := secrets.LoadMasterKey(getenv, secrets.MacKeychain) + if err != nil { + fmt.Fprintln(stderr, "current key:", err) + return 1 + } + next, err := secrets.ParseMasterKey(getenv("KAASU_NEW_MASTER_KEY")) + if err != nil { + fmt.Fprintln(stderr, "set KAASU_NEW_MASTER_KEY (make one with `kaasu keygen`):", err) + return 1 + } + st, err := store.Open(cfg.DatabasePath) + if err != nil { + fmt.Fprintln(stderr, "open database:", err) + return 1 + } + defer st.Close() + n, err := secrets.NewVault(st, cur, clock.Real{}).RotateAll(context.Background(), next) + if err != nil { + fmt.Fprintln(stderr, "rotate:", err) + return 1 + } + fmt.Fprintf(stdout, "rotated %d secrets; now point KAASU_MASTER_KEY at the new key\n", n) + return 0 +} + func serve(args []string, getenv func(string) string, stderr io.Writer) int { if len(args) > 0 { fmt.Fprintf(stderr, "serve: unexpected arguments %q (configure with KAASU_CONFIG or KAASU_* env)\n", args) diff --git a/cmd/kaasu/main_test.go b/cmd/kaasu/main_test.go index 7b69cc7..364cde5 100644 --- a/cmd/kaasu/main_test.go +++ b/cmd/kaasu/main_test.go @@ -56,3 +56,35 @@ func TestMigrateIsIdempotent(t *testing.T) { } } } + +func TestKeygenAndRotateKey(t *testing.T) { + dir := t.TempDir() + envs := map[string]string{"KAASU_DATABASE_PATH": filepath.Join(dir, "k.db")} + getenv := func(k string) string { return envs[k] } + + var out, errb bytes.Buffer + if code := run([]string{"keygen"}, getenv, &out, &errb); code != 0 { + t.Fatalf("keygen code=%d %q", code, errb.String()) + } + oldKey := strings.TrimSpace(out.String()) + out.Reset() + run([]string{"keygen"}, getenv, &out, &errb) + newKey := strings.TrimSpace(out.String()) + if oldKey == newKey || len(oldKey) < 40 { + t.Fatalf("keys must be random and long: %q %q", oldKey, newKey) + } + + envs["KAASU_MASTER_KEY"] = oldKey + if code := run([]string{"migrate"}, getenv, &out, &errb); code != 0 { + t.Fatalf("migrate: %q", errb.String()) + } + out.Reset() + if code := run([]string{"rotate-key"}, getenv, &out, &errb); code != 1 || !strings.Contains(errb.String(), "KAASU_NEW_MASTER_KEY") { + t.Errorf("rotate-key without a new key should explain: code=%d err=%q", code, errb.String()) + } + envs["KAASU_NEW_MASTER_KEY"] = newKey + errb.Reset() + if code := run([]string{"rotate-key"}, getenv, &out, &errb); code != 0 || !strings.Contains(out.String(), "rotated 0 secrets") { + t.Errorf("rotate-key: code=%d out=%q err=%q", code, out.String(), errb.String()) + } +} diff --git a/go.mod b/go.mod index cd4017b..d680919 100644 --- a/go.mod +++ b/go.mod @@ -8,6 +8,7 @@ require ( github.com/mattn/go-isatty v0.0.24 // indirect github.com/ncruces/go-strftime v1.1.0 // indirect github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect + golang.org/x/crypto v0.58.0 // indirect golang.org/x/sys v0.49.0 // indirect modernc.org/libc v1.79.0 // indirect modernc.org/mathutil v1.7.1 // indirect diff --git a/go.sum b/go.sum index 1e9e49e..c78c5c0 100644 --- a/go.sum +++ b/go.sum @@ -8,6 +8,8 @@ github.com/ncruces/go-strftime v1.1.0 h1:vtPmPWMkb539RGV9WpWT9EwJGvrOCKvMNI4obWL github.com/ncruces/go-strftime v1.1.0/go.mod h1:yVaq7iwCN09vHRrKBrrflXT0M9tTMG2KmZ9Fj7xeG/g= github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec h1:W09IVJc94icq4NjY3clb7Lk8O1qJ8BdBEF8z0ibU0rE= github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo= +golang.org/x/crypto v0.58.0 h1:COkYLr4k7nDI4r6QtMML2AfzeIswWHAJthLS6K/M/54= +golang.org/x/crypto v0.58.0/go.mod h1:Xh+kl5A+M0gMiWDytH2Hhr5Z7WbNCpv/eB/EFgfSk+w= golang.org/x/sys v0.49.0 h1:XbzkgYJHdqh/8m2Uu0W/dQv8nktxx4BFHp1M0gROTXA= golang.org/x/sys v0.49.0/go.mod h1:hNLxWAXmnKAxqDtdwIYC4bM9oQPEecfsnNMuSxOs3og= modernc.org/libc v1.79.0 h1:9Dvvv++bAMSkSuBzuOFAatbTV5OIoq6Y6FohmD3e7q8= diff --git a/internal/secrets/crypto.go b/internal/secrets/crypto.go new file mode 100644 index 0000000..4b37084 --- /dev/null +++ b/internal/secrets/crypto.go @@ -0,0 +1,175 @@ +// Package secrets protects stored secrets with envelope encryption and issues API keys. +// +// Each secret gets its own random data key. The data key is wrapped by the master key, which never +// lives in the database. Rotating the master key re-wraps data keys without touching the data. +package secrets + +import ( + "bytes" + "crypto/rand" + "crypto/sha256" + "encoding/base64" + "errors" + "fmt" + "strings" + + "golang.org/x/crypto/chacha20poly1305" +) + +const ( + blobPrefix = "v1." + keyIDLen = 8 +) + +// MasterKey is the 32-byte key that wraps data keys. +type MasterKey []byte + +// ParseMasterKey decodes a base64 (std or URL) 32-byte key. +func ParseMasterKey(s string) (MasterKey, error) { + s = strings.TrimSpace(s) + if s == "" { + return nil, errors.New("master key is empty") + } + b, err := base64.StdEncoding.DecodeString(s) + if err != nil { + b, err = base64.URLEncoding.DecodeString(s) + } + if err != nil { + return nil, fmt.Errorf("master key is not valid base64: %w", err) + } + if len(b) != chacha20poly1305.KeySize { + return nil, fmt.Errorf("master key must be %d bytes, got %d", chacha20poly1305.KeySize, len(b)) + } + return MasterKey(b), nil +} + +func (k MasterKey) id() []byte { + h := sha256.Sum256(append([]byte("kaasu-key-id:"), k...)) + return h[:keyIDLen] +} + +func (k MasterKey) aead() (cipherAEAD, error) { return chacha20poly1305.NewX(k) } + +type cipherAEAD interface { + Seal(dst, nonce, plaintext, additionalData []byte) []byte + Open(dst, nonce, ciphertext, additionalData []byte) ([]byte, error) + NonceSize() int + Overhead() int +} + +var dekAAD = []byte("kaasu:dek") + +// layout: keyID | wrapNonce | wrappedDEK | dataNonce | ciphertext +func random(n int) []byte { + b := make([]byte, n) + if _, err := rand.Read(b); err != nil { + panic("secrets: crypto/rand failed: " + err.Error()) + } + return b +} + +// Seal encrypts plaintext. aad (for example the secret's name) is bound to the ciphertext. +func (k MasterKey) Seal(plaintext, aad []byte) (string, error) { + kek, err := k.aead() + if err != nil { + return "", err + } + dek := random(chacha20poly1305.KeySize) + data, err := chacha20poly1305.NewX(dek) + if err != nil { + return "", err + } + wrapNonce, dataNonce := random(kek.NonceSize()), random(data.NonceSize()) + var out bytes.Buffer + out.Write(k.id()) + out.Write(wrapNonce) + out.Write(kek.Seal(nil, wrapNonce, dek, dekAAD)) + out.Write(dataNonce) + out.Write(data.Seal(nil, dataNonce, plaintext, aad)) + return blobPrefix + base64.RawURLEncoding.EncodeToString(out.Bytes()), nil +} + +type parsed struct { + keyID, wrapNonce, wrapped, dataNonce, ct []byte +} + +func parse(blob string) (parsed, error) { + var p parsed + if !strings.HasPrefix(blob, blobPrefix) { + return p, errors.New("not a kaasu secret blob") + } + raw, err := base64.RawURLEncoding.DecodeString(strings.TrimPrefix(blob, blobPrefix)) + if err != nil { + return p, fmt.Errorf("bad secret blob encoding: %w", err) + } + n := chacha20poly1305.NonceSizeX + wrapped := chacha20poly1305.KeySize + chacha20poly1305.Overhead + if len(raw) < keyIDLen+n+wrapped+n+chacha20poly1305.Overhead { + return p, errors.New("secret blob too short") + } + p.keyID, raw = raw[:keyIDLen], raw[keyIDLen:] + p.wrapNonce, raw = raw[:n], raw[n:] + p.wrapped, raw = raw[:wrapped], raw[wrapped:] + p.dataNonce, p.ct = raw[:n], raw[n:] + return p, nil +} + +func (k MasterKey) unwrap(p parsed) ([]byte, error) { + if !bytes.Equal(p.keyID, k.id()) { + return nil, errors.New("secret was sealed with a different master key") + } + kek, err := k.aead() + if err != nil { + return nil, err + } + dek, err := kek.Open(nil, p.wrapNonce, p.wrapped, dekAAD) + if err != nil { + return nil, errors.New("cannot unwrap data key (wrong master key or corrupted)") + } + return dek, nil +} + +// Open decrypts a blob made by Seal with the same aad. +func (k MasterKey) Open(blob string, aad []byte) ([]byte, error) { + p, err := parse(blob) + if err != nil { + return nil, err + } + dek, err := k.unwrap(p) + if err != nil { + return nil, err + } + data, err := chacha20poly1305.NewX(dek) + if err != nil { + return nil, err + } + pt, err := data.Open(nil, p.dataNonce, p.ct, aad) + if err != nil { + return nil, errors.New("cannot decrypt secret (tampered, or bound to another name)") + } + return pt, nil +} + +// Rewrap moves a blob to another master key without decrypting the data. +func (k MasterKey) Rewrap(blob string, to MasterKey) (string, error) { + p, err := parse(blob) + if err != nil { + return "", err + } + dek, err := k.unwrap(p) + if err != nil { + return "", err + } + kek, err := to.aead() + if err != nil { + return "", err + } + wrapNonce := random(kek.NonceSize()) + var out bytes.Buffer + out.Write(to.id()) + out.Write(wrapNonce) + out.Write(kek.Seal(nil, wrapNonce, dek, dekAAD)) + out.Write(p.dataNonce) + out.Write(p.ct) + return blobPrefix + base64.RawURLEncoding.EncodeToString(out.Bytes()), nil +} diff --git a/internal/secrets/key.go b/internal/secrets/key.go new file mode 100644 index 0000000..4e5adee --- /dev/null +++ b/internal/secrets/key.go @@ -0,0 +1,45 @@ +package secrets + +import ( + "errors" + "os" + "os/exec" + "path/filepath" + "runtime" + "strings" +) + +// LoadMasterKey finds the master key: KAASU_MASTER_KEY, then a systemd credential, then the OS keychain. +// It returns where the key came from so the operator can see it in the logs. +func LoadMasterKey(getenv func(string) string, keychain func() (string, error)) (MasterKey, string, error) { + if v := getenv("KAASU_MASTER_KEY"); v != "" { + k, err := ParseMasterKey(v) + return k, "env", err + } + if dir := getenv("CREDENTIALS_DIRECTORY"); dir != "" { + if b, err := os.ReadFile(filepath.Join(dir, "kaasu-master-key")); err == nil { + k, err := ParseMasterKey(string(b)) + return k, "systemd", err + } + } + if keychain != nil { + if v, err := keychain(); err == nil && v != "" { + k, err := ParseMasterKey(v) + return k, "keychain", err + } + } + return nil, "", errors.New("no master key found: set KAASU_MASTER_KEY, provide a systemd credential named kaasu-master-key, or store one in the OS keychain (generate with: kaasu keygen)") +} + +// MacKeychain reads the key from the macOS login keychain (service "kaasu", account "master-key"). +// On other systems it reports not found, so the other sources are used. +func MacKeychain() (string, error) { + if runtime.GOOS != "darwin" { + return "", os.ErrNotExist + } + out, err := exec.Command("/usr/bin/security", "find-generic-password", "-s", "kaasu", "-a", "master-key", "-w").Output() + if err != nil { + return "", err + } + return strings.TrimSpace(string(out)), nil +} diff --git a/internal/secrets/secrets_test.go b/internal/secrets/secrets_test.go new file mode 100644 index 0000000..7fd4a77 --- /dev/null +++ b/internal/secrets/secrets_test.go @@ -0,0 +1,203 @@ +package secrets + +import ( + "bytes" + "context" + "database/sql" + "encoding/base64" + "os" + "path/filepath" + "strings" + "testing" + "time" + + "github.com/opensawce/kaasu/internal/clock" + "github.com/opensawce/kaasu/internal/testutil" +) + +func key(b byte) MasterKey { return MasterKey(bytes.Repeat([]byte{b}, 32)) } + +func TestSealOpenRoundTrip(t *testing.T) { + k := key(1) + plain := []byte("imap-password-123") + a, err := k.Seal(plain, []byte("imap")) + if err != nil { + t.Fatal(err) + } + b, _ := k.Seal(plain, []byte("imap")) + if a == b { + t.Error("two seals of the same plaintext must differ (random nonce and data key)") + } + if strings.Contains(a, "imap-password") || strings.Contains(a, base64.StdEncoding.EncodeToString(plain)) { + t.Error("blob leaks plaintext") + } + got, err := k.Open(a, []byte("imap")) + if err != nil || !bytes.Equal(got, plain) { + t.Fatalf("got %q err=%v", got, err) + } +} + +func TestOpenRejectsWrongKeyAADAndTamper(t *testing.T) { + blob, _ := key(1).Seal([]byte("s3cret"), []byte("webhook")) + if _, err := key(2).Open(blob, []byte("webhook")); err == nil { + t.Error("wrong master key must fail") + } + if _, err := key(1).Open(blob, []byte("imap")); err == nil { + t.Error("wrong AAD (secret moved to another name) must fail") + } + raw, _ := base64.RawURLEncoding.DecodeString(strings.TrimPrefix(blob, "v1.")) + raw[len(raw)-1] ^= 1 + if _, err := key(1).Open("v1."+base64.RawURLEncoding.EncodeToString(raw), []byte("webhook")); err == nil { + t.Error("tampered ciphertext must fail") + } + if _, err := key(1).Open("garbage", nil); err == nil { + t.Error("garbage must fail") + } +} + +func TestRewrapMovesToNewKey(t *testing.T) { + old, nu := key(1), key(2) + blob, _ := old.Seal([]byte("ai-key"), []byte("ai")) + re, err := old.Rewrap(blob, nu) + if err != nil { + t.Fatal(err) + } + if got, err := nu.Open(re, []byte("ai")); err != nil || string(got) != "ai-key" { + t.Fatalf("new key cannot open: %v", err) + } + if _, err := old.Open(re, []byte("ai")); err == nil { + t.Error("old key must no longer open rewrapped blob") + } +} + +func TestParseMasterKey(t *testing.T) { + good := base64.StdEncoding.EncodeToString(bytes.Repeat([]byte{7}, 32)) + if _, err := ParseMasterKey(good); err != nil { + t.Errorf("good key rejected: %v", err) + } + for _, bad := range []string{"", "not base64!", base64.StdEncoding.EncodeToString([]byte("short"))} { + if _, err := ParseMasterKey(bad); err == nil { + t.Errorf("%q should be rejected", bad) + } + } +} + +func TestLoadMasterKeyPrecedence(t *testing.T) { + envKey := base64.StdEncoding.EncodeToString(bytes.Repeat([]byte{1}, 32)) + fileKey := base64.StdEncoding.EncodeToString(bytes.Repeat([]byte{2}, 32)) + kcKey := base64.StdEncoding.EncodeToString(bytes.Repeat([]byte{3}, 32)) + dir := t.TempDir() + if err := os.WriteFile(filepath.Join(dir, "kaasu-master-key"), []byte(fileKey+"\n"), 0o600); err != nil { + t.Fatal(err) + } + keychain := func() (string, error) { return kcKey, nil } + noKeychain := func() (string, error) { return "", os.ErrNotExist } + env := func(m map[string]string) func(string) string { return func(k string) string { return m[k] } } + + got, src, err := LoadMasterKey(env(map[string]string{"KAASU_MASTER_KEY": envKey, "CREDENTIALS_DIRECTORY": dir}), keychain) + if err != nil || src != "env" || got[0] != 1 { + t.Errorf("env should win: src=%s err=%v", src, err) + } + got, src, err = LoadMasterKey(env(map[string]string{"CREDENTIALS_DIRECTORY": dir}), keychain) + if err != nil || src != "systemd" || got[0] != 2 { + t.Errorf("systemd next: src=%s err=%v", src, err) + } + got, src, err = LoadMasterKey(env(nil), keychain) + if err != nil || src != "keychain" || got[0] != 3 { + t.Errorf("keychain last: src=%s err=%v", src, err) + } + _, _, err = LoadMasterKey(env(nil), noKeychain) + if err == nil || !strings.Contains(err.Error(), "KAASU_MASTER_KEY") { + t.Errorf("no key anywhere should explain the three sources, got %v", err) + } +} + +func newVault(t *testing.T, k MasterKey) *Vault { + t.Helper() + return NewVault(testutil.OpenStore(t), k, clock.NewFake(time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC))) +} + +func TestVaultStoresNoPlaintext(t *testing.T) { + v := newVault(t, key(1)) + ctx := context.Background() + if err := v.Put(ctx, "imap_password", []byte("hunter2")); err != nil { + t.Fatal(err) + } + var blob string + if err := v.st.Read().QueryRow(`SELECT blob FROM secrets WHERE name='imap_password'`).Scan(&blob); err != nil { + t.Fatal(err) + } + if strings.Contains(blob, "hunter2") { + t.Error("plaintext in database") + } + got, err := v.Get(ctx, "imap_password") + if err != nil || string(got) != "hunter2" { + t.Fatalf("got %q err=%v", got, err) + } + if _, err := v.Get(ctx, "missing"); err != ErrNotFound { + t.Errorf("want ErrNotFound, got %v", err) + } + if err := v.Put(ctx, "imap_password", []byte("new")); err != nil { + t.Fatal(err) + } + if got, _ := v.Get(ctx, "imap_password"); string(got) != "new" { + t.Error("put should overwrite") + } +} + +func TestRotateAllIsAtomic(t *testing.T) { + v := newVault(t, key(1)) + ctx := context.Background() + for _, n := range []string{"a", "b", "c"} { + if err := v.Put(ctx, n, []byte("val-"+n)); err != nil { + t.Fatal(err) + } + } + // Corrupt one row: rotation must fail and change nothing. + _ = v.st.Tx(ctx, func(tx *sql.Tx) error { _, e := tx.Exec(`UPDATE secrets SET blob='v1.AAAA' WHERE name='b'`); return e }) + if n, err := v.RotateAll(ctx, key(2)); err == nil || n != 0 { + t.Fatalf("rotation over a corrupt row should fail with 0 changed, got n=%d err=%v", n, err) + } + if got, err := v.Get(ctx, "a"); err != nil || string(got) != "val-a" { + t.Errorf("row a changed by failed rotation: %q %v", got, err) + } + // Repair and rotate for real. + _ = v.Put(ctx, "b", []byte("val-b")) + n, err := v.RotateAll(ctx, key(2)) + if err != nil || n != 3 { + t.Fatalf("n=%d err=%v", n, err) + } + if got, err := v.Get(ctx, "c"); err != nil || string(got) != "val-c" { + t.Errorf("after rotation Get uses new key: %q %v", got, err) + } +} + +func TestAPIKeys(t *testing.T) { + v := newVault(t, key(1)) + ctx := context.Background() + plain, rec, err := v.NewAPIKey(ctx, "WooCommerce") + if err != nil { + t.Fatal(err) + } + if !strings.HasPrefix(plain, "kaasu_live_") || len(plain) < 40 { + t.Errorf("odd key %q", plain) + } + var hash string + _ = v.st.Read().QueryRow(`SELECT key_hash FROM api_keys WHERE id=?`, rec.ID).Scan(&hash) + if hash == "" || strings.Contains(hash, plain) || len(hash) != 64 { + t.Errorf("must store a sha-256 hex hash, got %q", hash) + } + got, err := v.VerifyAPIKey(ctx, plain) + if err != nil || got.ID != rec.ID || got.Label != "WooCommerce" { + t.Fatalf("verify: %+v %v", got, err) + } + if _, err := v.VerifyAPIKey(ctx, plain+"x"); err != ErrInvalidKey { + t.Errorf("wrong key: %v", err) + } + if err := v.RevokeAPIKey(ctx, rec.ID); err != nil { + t.Fatal(err) + } + if _, err := v.VerifyAPIKey(ctx, plain); err != ErrInvalidKey { + t.Errorf("revoked key must fail: %v", err) + } +} diff --git a/internal/secrets/vault.go b/internal/secrets/vault.go new file mode 100644 index 0000000..e21d803 --- /dev/null +++ b/internal/secrets/vault.go @@ -0,0 +1,175 @@ +package secrets + +import ( + "context" + "crypto/rand" + "crypto/sha256" + "database/sql" + "encoding/base32" + "encoding/hex" + "errors" + "fmt" + "strings" + "sync" + "time" + + "github.com/opensawce/kaasu/internal/clock" + "github.com/opensawce/kaasu/internal/id" + "github.com/opensawce/kaasu/internal/store" +) + +var ( + ErrNotFound = errors.New("secret not found") + ErrInvalidKey = errors.New("invalid or revoked API key") +) + +// Vault stores encrypted secrets and API key hashes in the database. +type Vault struct { + st *store.Store + clk clock.Clock + mu sync.RWMutex + key MasterKey +} + +func NewVault(st *store.Store, key MasterKey, clk clock.Clock) *Vault { + return &Vault{st: st, key: key, clk: clk} +} + +func (v *Vault) masterKey() MasterKey { + v.mu.RLock() + defer v.mu.RUnlock() + return v.key +} + +func (v *Vault) now() string { return v.clk.Now().UTC().Format(time.RFC3339) } + +// Put encrypts and stores a secret under name, replacing any previous value. +func (v *Vault) Put(ctx context.Context, name string, plaintext []byte) error { + blob, err := v.masterKey().Seal(plaintext, []byte(name)) + if err != nil { + return fmt.Errorf("seal %s: %w", name, err) + } + return v.st.Tx(ctx, func(tx *sql.Tx) error { + _, err := tx.ExecContext(ctx, `INSERT INTO secrets(name,blob,updated_at) VALUES(?,?,?) + ON CONFLICT(name) DO UPDATE SET blob=excluded.blob, updated_at=excluded.updated_at`, name, blob, v.now()) + return err + }) +} + +// Get decrypts the secret stored under name. +func (v *Vault) Get(ctx context.Context, name string) ([]byte, error) { + var blob string + err := v.st.Read().QueryRowContext(ctx, `SELECT blob FROM secrets WHERE name = ?`, name).Scan(&blob) + if errors.Is(err, sql.ErrNoRows) { + return nil, ErrNotFound + } + if err != nil { + return nil, err + } + return v.masterKey().Open(blob, []byte(name)) +} + +// RotateAll re-wraps every secret under next in one transaction. It changes nothing if any row fails. +func (v *Vault) RotateAll(ctx context.Context, next MasterKey) (int, error) { + cur := v.masterKey() + count := 0 + err := v.st.Tx(ctx, func(tx *sql.Tx) error { + rows, err := tx.QueryContext(ctx, `SELECT name, blob FROM secrets`) + if err != nil { + return err + } + type row struct{ name, blob string } + var all []row + for rows.Next() { + var r row + if err := rows.Scan(&r.name, &r.blob); err != nil { + _ = rows.Close() + return err + } + all = append(all, r) + } + if err := rows.Err(); err != nil { + _ = rows.Close() + return err + } + if err := rows.Close(); err != nil { + return err + } + for _, r := range all { + nb, err := cur.Rewrap(r.blob, next) + if err != nil { + return fmt.Errorf("rewrap %s: %w", r.name, err) + } + if _, err := tx.ExecContext(ctx, `UPDATE secrets SET blob=?, updated_at=? WHERE name=?`, nb, v.now(), r.name); err != nil { + return err + } + } + count = len(all) + return nil + }) + if err != nil { + return 0, err + } + v.mu.Lock() + v.key = next + v.mu.Unlock() + return count, nil +} + +// APIKey is the stored record. The key itself is never stored and cannot be shown again. +type APIKey struct { + ID, Label, Prefix string +} + +var b32 = base32.NewEncoding("abcdefghijklmnopqrstuvwxyz234567").WithPadding(base32.NoPadding) + +func hashKey(k string) string { + h := sha256.Sum256([]byte(k)) + return hex.EncodeToString(h[:]) +} + +// NewAPIKey creates a key and returns it once. Only its SHA-256 hash is stored. +func (v *Vault) NewAPIKey(ctx context.Context, label string) (string, APIKey, error) { + raw := make([]byte, 32) + if _, err := rand.Read(raw); err != nil { + return "", APIKey{}, err + } + plain := "kaasu_live_" + b32.EncodeToString(raw) + rec := APIKey{ID: id.New("key"), Label: label, Prefix: plain[:15]} + err := v.st.Tx(ctx, func(tx *sql.Tx) error { + _, err := tx.ExecContext(ctx, `INSERT INTO api_keys(id,label,prefix,key_hash,created_at) VALUES(?,?,?,?,?)`, + rec.ID, rec.Label, rec.Prefix, hashKey(plain), v.now()) + return err + }) + if err != nil { + return "", APIKey{}, err + } + return plain, rec, nil +} + +// VerifyAPIKey looks the key up by its hash (a 256-bit hash, so lookup timing reveals nothing usable). +func (v *Vault) VerifyAPIKey(ctx context.Context, plain string) (APIKey, error) { + if !strings.HasPrefix(plain, "kaasu_live_") { + return APIKey{}, ErrInvalidKey + } + var k APIKey + err := v.st.Read().QueryRowContext(ctx, + `SELECT id,label,prefix FROM api_keys WHERE key_hash = ? AND revoked_at IS NULL`, hashKey(plain)).Scan(&k.ID, &k.Label, &k.Prefix) + if errors.Is(err, sql.ErrNoRows) { + return APIKey{}, ErrInvalidKey + } + return k, err +} + +func (v *Vault) RevokeAPIKey(ctx context.Context, keyID string) error { + return v.st.Tx(ctx, func(tx *sql.Tx) error { + res, err := tx.ExecContext(ctx, `UPDATE api_keys SET revoked_at=? WHERE id=? AND revoked_at IS NULL`, v.now(), keyID) + if err != nil { + return err + } + if n, _ := res.RowsAffected(); n == 0 { + return ErrInvalidKey + } + return nil + }) +} diff --git a/internal/store/migrations/0002_secrets.sql b/internal/store/migrations/0002_secrets.sql new file mode 100644 index 0000000..7eb6432 --- /dev/null +++ b/internal/store/migrations/0002_secrets.sql @@ -0,0 +1,14 @@ +CREATE TABLE secrets ( + name TEXT PRIMARY KEY, + blob TEXT NOT NULL, + updated_at TEXT NOT NULL +) STRICT; + +CREATE TABLE api_keys ( + id TEXT PRIMARY KEY, + label TEXT NOT NULL, + prefix TEXT NOT NULL, + key_hash TEXT NOT NULL UNIQUE, + created_at TEXT NOT NULL, + revoked_at TEXT +) STRICT; From d724237b1fef9c0afb2784bdc101783262c5e7be Mon Sep 17 00:00:00 2001 From: Manish <69393822+manish-9245@users.noreply.github.com> Date: Sat, 24 Oct 2026 12:18:37 +0530 Subject: [PATCH 07/54] feat(server): receiving accounts (multiple UPI IDs per shop) (#92) * feat(server): receiving accounts * fix: microsecond timestamps keep account list order * test: expect microsecond timestamp --- internal/accounts/accounts.go | 188 +++++++++++++++++++++++++++++ internal/accounts/accounts_test.go | 112 +++++++++++++++++ 2 files changed, 300 insertions(+) create mode 100644 internal/accounts/accounts.go create mode 100644 internal/accounts/accounts_test.go diff --git a/internal/accounts/accounts.go b/internal/accounts/accounts.go new file mode 100644 index 0000000..99a3470 --- /dev/null +++ b/internal/accounts/accounts.go @@ -0,0 +1,188 @@ +// Package accounts manages the UPI IDs a shop receives money on. +package accounts + +import ( + "context" + "database/sql" + "errors" + "fmt" + "regexp" + "strings" + + "github.com/opensawce/kaasu/internal/clock" + "github.com/opensawce/kaasu/internal/store" +) + +var ( + ErrNotFound = errors.New("receiving account not found") + ErrInvalid = errors.New("invalid receiving account") + ErrDuplicate = errors.New("a receiving account with this UPI ID already exists") + ErrInUse = errors.New("receiving account has orders; deactivate it instead") +) + +var vpaRe = regexp.MustCompile(`^[a-z0-9][a-z0-9._-]{0,255}@[a-z][a-z0-9]{2,63}$`) + +// IDSource makes ids; testutil.Seq and a random source both satisfy it. +type IDSource interface{ New(prefix string) string } + +// Account is one receiving UPI ID. VPAs are stored lower-case. +type Account struct { + ID string + VPA string + Bank string + DisplayName string + AutoConfirmLimitPaise int64 // orders above this always need the owner's OK + Active bool + CreatedAt string +} + +// Input is what the owner provides when creating or editing an account. +type Input struct { + VPA string + Bank string + DisplayName string + AutoConfirmLimitPaise int64 +} + +type Service struct { + st *store.Store + clk clock.Clock + ids IDSource +} + +func New(st *store.Store, clk clock.Clock, ids IDSource) *Service { + return &Service{st: st, clk: clk, ids: ids} +} + +func (in *Input) normalise() error { + in.VPA = strings.ToLower(strings.TrimSpace(in.VPA)) + in.Bank = strings.TrimSpace(in.Bank) + in.DisplayName = strings.TrimSpace(in.DisplayName) + switch { + case !vpaRe.MatchString(in.VPA): + return fmt.Errorf("%w: UPI ID %q is not name@handle", ErrInvalid, in.VPA) + case in.Bank == "": + return fmt.Errorf("%w: bank is required", ErrInvalid) + case in.DisplayName == "": + return fmt.Errorf("%w: display name is required", ErrInvalid) + case in.AutoConfirmLimitPaise < 0: + return fmt.Errorf("%w: auto-confirm limit cannot be negative", ErrInvalid) + } + return nil +} + +const cols = `id, vpa, bank, display_name, auto_confirm_limit_paise, active, created_at` + +func scan(r interface{ Scan(...any) error }) (Account, error) { + var a Account + var active int + err := r.Scan(&a.ID, &a.VPA, &a.Bank, &a.DisplayName, &a.AutoConfirmLimitPaise, &active, &a.CreatedAt) + a.Active = active == 1 + return a, err +} + +func isUnique(err error) bool { + return err != nil && strings.Contains(err.Error(), "UNIQUE constraint failed") +} + +func (s *Service) Create(ctx context.Context, in Input) (Account, error) { + if err := in.normalise(); err != nil { + return Account{}, err + } + a := Account{ID: s.ids.New("acc"), VPA: in.VPA, Bank: in.Bank, DisplayName: in.DisplayName, + AutoConfirmLimitPaise: in.AutoConfirmLimitPaise, Active: true, CreatedAt: s.clk.Now().UTC().Format("2006-01-02T15:04:05.000000Z")} + err := s.st.Tx(ctx, func(tx *sql.Tx) error { + _, err := tx.ExecContext(ctx, `INSERT INTO receiving_accounts(`+cols+`) VALUES(?,?,?,?,?,1,?)`, + a.ID, a.VPA, a.Bank, a.DisplayName, a.AutoConfirmLimitPaise, a.CreatedAt) + return err + }) + if isUnique(err) { + return Account{}, ErrDuplicate + } + return a, err +} + +func (s *Service) Get(ctx context.Context, id string) (Account, error) { + a, err := scan(s.st.Read().QueryRowContext(ctx, `SELECT `+cols+` FROM receiving_accounts WHERE id = ?`, id)) + if errors.Is(err, sql.ErrNoRows) { + return Account{}, ErrNotFound + } + return a, err +} + +// List returns active accounts first, then inactive, oldest first within each group. +func (s *Service) List(ctx context.Context) ([]Account, error) { + rows, err := s.st.Read().QueryContext(ctx, `SELECT `+cols+` FROM receiving_accounts ORDER BY active DESC, created_at, id`) + if err != nil { + return nil, err + } + defer rows.Close() + var out []Account + for rows.Next() { + a, err := scan(rows) + if err != nil { + return nil, err + } + out = append(out, a) + } + return out, rows.Err() +} + +func (s *Service) Update(ctx context.Context, id string, in Input) (Account, error) { + if err := in.normalise(); err != nil { + return Account{}, err + } + err := s.st.Tx(ctx, func(tx *sql.Tx) error { + res, err := tx.ExecContext(ctx, `UPDATE receiving_accounts SET vpa=?, bank=?, display_name=?, auto_confirm_limit_paise=? WHERE id=?`, + in.VPA, in.Bank, in.DisplayName, in.AutoConfirmLimitPaise, id) + if err != nil { + return err + } + if n, _ := res.RowsAffected(); n == 0 { + return ErrNotFound + } + return nil + }) + if isUnique(err) { + return Account{}, ErrDuplicate + } + if err != nil { + return Account{}, err + } + return s.Get(ctx, id) +} + +// Deactivate stops new orders on the account; existing orders keep working. +func (s *Service) Deactivate(ctx context.Context, id string) error { + return s.st.Tx(ctx, func(tx *sql.Tx) error { + res, err := tx.ExecContext(ctx, `UPDATE receiving_accounts SET active=0 WHERE id=?`, id) + if err != nil { + return err + } + if n, _ := res.RowsAffected(); n == 0 { + return ErrNotFound + } + return nil + }) +} + +// Delete removes an account that never had orders. +func (s *Service) Delete(ctx context.Context, id string) error { + return s.st.Tx(ctx, func(tx *sql.Tx) error { + var n int + if err := tx.QueryRowContext(ctx, `SELECT count(*) FROM orders WHERE account_id = ?`, id).Scan(&n); err != nil { + return err + } + if n > 0 { + return ErrInUse + } + res, err := tx.ExecContext(ctx, `DELETE FROM receiving_accounts WHERE id=?`, id) + if err != nil { + return err + } + if c, _ := res.RowsAffected(); c == 0 { + return ErrNotFound + } + return nil + }) +} diff --git a/internal/accounts/accounts_test.go b/internal/accounts/accounts_test.go new file mode 100644 index 0000000..03d654b --- /dev/null +++ b/internal/accounts/accounts_test.go @@ -0,0 +1,112 @@ +package accounts + +import ( + "context" + "database/sql" + "errors" + "testing" + "time" + + "github.com/opensawce/kaasu/internal/clock" + "github.com/opensawce/kaasu/internal/testutil" +) + +func newSvc(t *testing.T) *Service { + t.Helper() + return New(testutil.OpenStore(t), clock.NewFake(time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC)), &testutil.Seq{}) +} + +func TestCreateAndGet(t *testing.T) { + s := newSvc(t) + ctx := context.Background() + a, err := s.Create(ctx, Input{VPA: "sharma.sweets@okhdfcbank", Bank: "HDFC", DisplayName: "Sharma Sweets", AutoConfirmLimitPaise: 500000}) + if err != nil { + t.Fatal(err) + } + if a.ID != "acc_000001" || !a.Active || a.CreatedAt != "2026-01-01T00:00:00.000000Z" { + t.Errorf("unexpected account %+v", a) + } + got, err := s.Get(ctx, a.ID) + if err != nil || got != a { + t.Fatalf("get=%+v err=%v", got, err) + } + if _, err := s.Get(ctx, "acc_nope"); !errors.Is(err, ErrNotFound) { + t.Errorf("want ErrNotFound, got %v", err) + } +} + +func TestValidation(t *testing.T) { + s := newSvc(t) + good := Input{VPA: "shop@okbank", Bank: "Bank", DisplayName: "Shop"} + tests := []struct { + name string + mut func(*Input) + }{ + {"no at sign", func(i *Input) { i.VPA = "shopokbank" }}, + {"space in vpa", func(i *Input) { i.VPA = "my shop@okbank" }}, + {"handle too short", func(i *Input) { i.VPA = "shop@ab" }}, + {"empty name part", func(i *Input) { i.VPA = "@okbank" }}, + {"two at signs", func(i *Input) { i.VPA = "a@b@okbank" }}, + {"missing bank", func(i *Input) { i.Bank = " " }}, + {"missing display name", func(i *Input) { i.DisplayName = "" }}, + {"negative limit", func(i *Input) { i.AutoConfirmLimitPaise = -1 }}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + in := good + tt.mut(&in) + if _, err := s.Create(context.Background(), in); !errors.Is(err, ErrInvalid) { + t.Errorf("want ErrInvalid, got %v", err) + } + }) + } +} + +func TestDuplicateVPAIsCaseInsensitive(t *testing.T) { + s := newSvc(t) + ctx := context.Background() + if _, err := s.Create(ctx, Input{VPA: "shop@okbank", Bank: "B", DisplayName: "S"}); err != nil { + t.Fatal(err) + } + if _, err := s.Create(ctx, Input{VPA: "SHOP@OkBank", Bank: "B", DisplayName: "S2"}); !errors.Is(err, ErrDuplicate) { + t.Errorf("want ErrDuplicate, got %v", err) + } +} + +func TestUpdateListDeactivate(t *testing.T) { + s := newSvc(t) + ctx := context.Background() + a, _ := s.Create(ctx, Input{VPA: "a@okbank", Bank: "B", DisplayName: "A"}) + b, _ := s.Create(ctx, Input{VPA: "b@okbank", Bank: "B", DisplayName: "B"}) + up, err := s.Update(ctx, a.ID, Input{VPA: "a@okbank", Bank: "ICICI", DisplayName: "A shop", AutoConfirmLimitPaise: 100}) + if err != nil || up.Bank != "ICICI" || up.AutoConfirmLimitPaise != 100 { + t.Fatalf("update: %+v %v", up, err) + } + if err := s.Deactivate(ctx, a.ID); err != nil { + t.Fatal(err) + } + list, err := s.List(ctx) + if err != nil || len(list) != 2 || list[0].ID != b.ID || list[1].Active { + t.Errorf("active accounts should list first: %+v %v", list, err) + } +} + +func TestDeleteBlockedWhenOrdersExist(t *testing.T) { + s := newSvc(t) + ctx := context.Background() + a, _ := s.Create(ctx, Input{VPA: "a@okbank", Bank: "B", DisplayName: "A"}) + if err := s.st.Tx(ctx, func(tx *sql.Tx) error { + _, e := tx.Exec(`INSERT INTO orders(id,account_id,base_amount_paise,amount_paise,state,expires_at,created_at,updated_at) + VALUES('o1',?,100,100,'awaiting','2026-01-01T00:15:00Z','2026-01-01T00:00:00Z','2026-01-01T00:00:00Z')`, a.ID) + return e + }); err != nil { + t.Fatal(err) + } + if err := s.Delete(ctx, a.ID); !errors.Is(err, ErrInUse) { + t.Errorf("want ErrInUse, got %v", err) + } + b, _ := s.Create(ctx, Input{VPA: "b@okbank", Bank: "B", DisplayName: "B"}) + if err := s.Delete(ctx, b.ID); err != nil { + t.Errorf("unused account should delete: %v", err) + } +} From d8061c622cf52d7c5b162a626b4d74933301863f Mon Sep 17 00:00:00 2001 From: Manish <69393822+manish-9245@users.noreply.github.com> Date: Sat, 24 Oct 2026 12:22:29 +0530 Subject: [PATCH 08/54] feat(security): hash-chained ledger and kaasu verify (#93) * feat(security): hash-chained append-only ledger and kaasu verify * fix: reject separator byte in ledger fields; document truncation limit * test: cover separator guard and Describe so the ledger clears the 85% gate --- architecture.md | 2 +- cmd/kaasu/main.go | 34 ++- cmd/kaasu/main_test.go | 47 +++++ internal/accounts/accounts.go | 22 +- internal/accounts/accounts_test.go | 26 +++ internal/ledger/ledger.go | 143 +++++++++++++ internal/ledger/ledger_test.go | 199 ++++++++++++++++++ internal/secrets/secrets_test.go | 28 +++ internal/secrets/vault.go | 21 +- .../migrations/0003_ledger_append_only.sql | 7 + 10 files changed, 519 insertions(+), 10 deletions(-) create mode 100644 internal/ledger/ledger.go create mode 100644 internal/ledger/ledger_test.go create mode 100644 internal/store/migrations/0003_ledger_append_only.sql diff --git a/architecture.md b/architecture.md index d763b1f..244ec45 100644 --- a/architecture.md +++ b/architecture.md @@ -33,7 +33,7 @@ SQLite in WAL mode, STRICT tables, foreign keys on, one writer (`BEGIN IMMEDIATE | `evidence`, `alerts` | raw alert bodies and parsed facts | unique raw hash, unique UTR | | `matches` | alert to order decisions | one match per alert | | `refunds` | refunds recorded against orders | foreign key to orders | -| `ledger` | append-only hash chain | see ledger rules | +| `ledger` | append-only hash chain: `hash = SHA-256(prev_hash, seq, ts, kind, order, actor, payload)`; triggers block UPDATE and DELETE; `kaasu verify` reports the first broken row; the chain detects edited rows but not removal of the newest rows unless the latest hash is kept outside the database (see the backup work) | triggers plus the chain | | `outbox` | webhook events waiting for delivery | unique event id | ### Deployment diff --git a/cmd/kaasu/main.go b/cmd/kaasu/main.go index 9017b62..db8b1c2 100644 --- a/cmd/kaasu/main.go +++ b/cmd/kaasu/main.go @@ -15,6 +15,7 @@ import ( "github.com/opensawce/kaasu/internal/clock" "github.com/opensawce/kaasu/internal/config" + "github.com/opensawce/kaasu/internal/ledger" "github.com/opensawce/kaasu/internal/secrets" "github.com/opensawce/kaasu/internal/server" "github.com/opensawce/kaasu/internal/store" @@ -52,7 +53,9 @@ func run(args []string, getenv func(string) string, stdout, stderr io.Writer) in return 0 case "rotate-key": return rotateKey(getenv, stdout, stderr) - case "verify", "backup", "pair": + case "verify": + return verify(getenv, stdout, stderr) + case "backup", "pair": fmt.Fprintf(stderr, "kaasu %s: not implemented yet\n", args[0]) return 1 default: @@ -114,6 +117,35 @@ func rotateKey(getenv func(string) string, stdout, stderr io.Writer) int { return 0 } +// verify walks the ledger hash chain and exits non-zero at the first broken row. +func verify(getenv func(string) string, stdout, stderr io.Writer) int { + cfg, err := config.Load(getenv, getenv("KAASU_CONFIG")) + if err != nil { + fmt.Fprintln(stderr, "config:", err) + return 1 + } + st, err := store.Open(cfg.DatabasePath) + if err != nil { + fmt.Fprintln(stderr, "open database:", err) + return 1 + } + defer st.Close() + if err := st.Migrate(context.Background()); err != nil { + fmt.Fprintln(stderr, "migrate:", err) + return 1 + } + res, err := ledger.Verify(context.Background(), st.Read()) + if err != nil { + fmt.Fprintln(stderr, "verify:", err) + return 1 + } + fmt.Fprintln(stdout, res.Describe()) + if !res.OK { + return 1 + } + return 0 +} + func serve(args []string, getenv func(string) string, stderr io.Writer) int { if len(args) > 0 { fmt.Fprintf(stderr, "serve: unexpected arguments %q (configure with KAASU_CONFIG or KAASU_* env)\n", args) diff --git a/cmd/kaasu/main_test.go b/cmd/kaasu/main_test.go index 364cde5..aa92eaf 100644 --- a/cmd/kaasu/main_test.go +++ b/cmd/kaasu/main_test.go @@ -2,9 +2,16 @@ package main import ( "bytes" + "context" + "database/sql" "path/filepath" "strings" "testing" + "time" + + "github.com/opensawce/kaasu/internal/clock" + "github.com/opensawce/kaasu/internal/ledger" + "github.com/opensawce/kaasu/internal/store" ) func TestRun(t *testing.T) { @@ -88,3 +95,43 @@ func TestKeygenAndRotateKey(t *testing.T) { t.Errorf("rotate-key: code=%d out=%q err=%q", code, out.String(), errb.String()) } } + +func TestVerifyDetectsTampering(t *testing.T) { + db := filepath.Join(t.TempDir(), "k.db") + getenv := func(k string) string { + if k == "KAASU_DATABASE_PATH" { + return db + } + return "" + } + var out, errb bytes.Buffer + if code := run([]string{"verify"}, getenv, &out, &errb); code != 0 || !strings.Contains(out.String(), "ledger intact: 0 rows") { + t.Fatalf("fresh db: code=%d out=%q err=%q", code, out.String(), errb.String()) + } + st, err := store.Open(db) + if err != nil { + t.Fatal(err) + } + clk := clock.NewFake(time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC)) + for i := 0; i < 3; i++ { + _ = st.Tx(context.Background(), func(tx *sql.Tx) error { + _, err := ledger.Append(context.Background(), tx, clk, ledger.Entry{Kind: "order.created", OrderID: "ord_1"}) + return err + }) + } + out.Reset() + if code := run([]string{"verify"}, getenv, &out, &errb); code != 0 || !strings.Contains(out.String(), "3 rows") { + t.Fatalf("clean ledger: code=%d out=%q", code, out.String()) + } + _ = st.Tx(context.Background(), func(tx *sql.Tx) error { + _, _ = tx.Exec(`DROP TRIGGER ledger_no_update`) + _, err := tx.Exec(`UPDATE ledger SET actor='mallory' WHERE seq=2`) + return err + }) + st.Close() + out.Reset() + code := run([]string{"verify"}, getenv, &out, &errb) + if code != 1 || !strings.Contains(out.String(), "BROKEN at row 2") { + t.Errorf("tampered: code=%d out=%q", code, out.String()) + } +} diff --git a/internal/accounts/accounts.go b/internal/accounts/accounts.go index 99a3470..cdc28d7 100644 --- a/internal/accounts/accounts.go +++ b/internal/accounts/accounts.go @@ -10,6 +10,7 @@ import ( "strings" "github.com/opensawce/kaasu/internal/clock" + "github.com/opensawce/kaasu/internal/ledger" "github.com/opensawce/kaasu/internal/store" ) @@ -54,6 +55,16 @@ func New(st *store.Store, clk clock.Clock, ids IDSource) *Service { return &Service{st: st, clk: clk, ids: ids} } +// record writes the admin action to the ledger in the same transaction. Ledger rows have no order, so the account id goes in the payload. +func (s *Service) record(ctx context.Context, tx *sql.Tx, kind, accountID string, extra map[string]any) error { + p := map[string]any{"account_id": accountID} + for k, v := range extra { + p[k] = v + } + _, err := ledger.Append(ctx, tx, s.clk, ledger.Entry{Kind: kind, Payload: p}) + return err +} + func (in *Input) normalise() error { in.VPA = strings.ToLower(strings.TrimSpace(in.VPA)) in.Bank = strings.TrimSpace(in.Bank) @@ -94,7 +105,10 @@ func (s *Service) Create(ctx context.Context, in Input) (Account, error) { err := s.st.Tx(ctx, func(tx *sql.Tx) error { _, err := tx.ExecContext(ctx, `INSERT INTO receiving_accounts(`+cols+`) VALUES(?,?,?,?,?,1,?)`, a.ID, a.VPA, a.Bank, a.DisplayName, a.AutoConfirmLimitPaise, a.CreatedAt) - return err + if err != nil { + return err + } + return s.record(ctx, tx, "account.created", a.ID, map[string]any{"vpa": a.VPA, "bank": a.Bank, "auto_confirm_limit_paise": a.AutoConfirmLimitPaise}) }) if isUnique(err) { return Account{}, ErrDuplicate @@ -141,7 +155,7 @@ func (s *Service) Update(ctx context.Context, id string, in Input) (Account, err if n, _ := res.RowsAffected(); n == 0 { return ErrNotFound } - return nil + return s.record(ctx, tx, "account.updated", id, map[string]any{"vpa": in.VPA, "bank": in.Bank, "auto_confirm_limit_paise": in.AutoConfirmLimitPaise}) }) if isUnique(err) { return Account{}, ErrDuplicate @@ -162,7 +176,7 @@ func (s *Service) Deactivate(ctx context.Context, id string) error { if n, _ := res.RowsAffected(); n == 0 { return ErrNotFound } - return nil + return s.record(ctx, tx, "account.deactivated", id, nil) }) } @@ -183,6 +197,6 @@ func (s *Service) Delete(ctx context.Context, id string) error { if c, _ := res.RowsAffected(); c == 0 { return ErrNotFound } - return nil + return s.record(ctx, tx, "account.deleted", id, nil) }) } diff --git a/internal/accounts/accounts_test.go b/internal/accounts/accounts_test.go index 03d654b..92d2aa1 100644 --- a/internal/accounts/accounts_test.go +++ b/internal/accounts/accounts_test.go @@ -4,10 +4,12 @@ import ( "context" "database/sql" "errors" + "fmt" "testing" "time" "github.com/opensawce/kaasu/internal/clock" + "github.com/opensawce/kaasu/internal/ledger" "github.com/opensawce/kaasu/internal/testutil" ) @@ -110,3 +112,27 @@ func TestDeleteBlockedWhenOrdersExist(t *testing.T) { t.Errorf("unused account should delete: %v", err) } } + +func TestAdminChangesAreLedgered(t *testing.T) { + s := newSvc(t) + ctx := ledger.WithActor(context.Background(), "admin:priya") + a, _ := s.Create(ctx, Input{VPA: "a@okbank", Bank: "B", DisplayName: "A"}) + _, _ = s.Update(ctx, a.ID, Input{VPA: "a@okbank", Bank: "B2", DisplayName: "A"}) + _ = s.Deactivate(ctx, a.ID) + _ = s.Delete(ctx, a.ID) + rows, err := s.st.Read().Query(`SELECT kind, actor FROM ledger ORDER BY seq`) + if err != nil { + t.Fatal(err) + } + defer rows.Close() + var got []string + for rows.Next() { + var k, a string + _ = rows.Scan(&k, &a) + got = append(got, k+"/"+a) + } + want := []string{"account.created/admin:priya", "account.updated/admin:priya", "account.deactivated/admin:priya", "account.deleted/admin:priya"} + if fmt.Sprint(got) != fmt.Sprint(want) { + t.Errorf("got %v want %v", got, want) + } +} diff --git a/internal/ledger/ledger.go b/internal/ledger/ledger.go new file mode 100644 index 0000000..a0f8770 --- /dev/null +++ b/internal/ledger/ledger.go @@ -0,0 +1,143 @@ +// Package ledger is the append-only, hash-chained record of every money-relevant event. +// +// Each row's hash covers the previous row's hash, so changing or removing any row breaks every row after it. +// Append must be called inside the same transaction as the change it records. +package ledger + +import ( + "context" + "crypto/sha256" + "database/sql" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "strconv" + "strings" + "time" + + "github.com/opensawce/kaasu/internal/clock" +) + +const genesis = "0000000000000000000000000000000000000000000000000000000000000000" + +type actorKey struct{} + +// WithActor records who is acting ("admin:priya", "device:dev_1", "api:key_3"). The default is "system". +func WithActor(ctx context.Context, actor string) context.Context { + return context.WithValue(ctx, actorKey{}, actor) +} + +func actorFrom(ctx context.Context) string { + if a, ok := ctx.Value(actorKey{}).(string); ok && a != "" { + return a + } + return "system" +} + +// Entry is one event. Payload must not contain secrets or full payer identifiers. +type Entry struct { + Kind string + OrderID string + Actor string // optional; defaults to the context actor + Payload map[string]any +} + +func rowHash(prev string, seq int64, ts, kind, orderID, actor, payload string) string { + h := sha256.New() + for _, part := range []string{prev, strconv.FormatInt(seq, 10), ts, kind, orderID, actor, payload} { + h.Write([]byte(part)) + h.Write([]byte{0x1f}) // unit separator keeps field boundaries unambiguous + } + return hex.EncodeToString(h.Sum(nil)) +} + +// Append adds a row inside tx and returns its sequence number. +func Append(ctx context.Context, tx *sql.Tx, clk clock.Clock, e Entry) (int64, error) { + if e.Kind == "" { + return 0, errors.New("ledger: entry needs a kind") + } + actor := e.Actor + if actor == "" { + actor = actorFrom(ctx) + } + for _, f := range []string{e.Kind, e.OrderID, actor} { + if strings.ContainsRune(f, 0x1f) { + return 0, errors.New("ledger: field contains the hash separator byte") + } + } + payload := "{}" + if len(e.Payload) > 0 { + b, err := json.Marshal(e.Payload) // map keys are sorted, so the text is stable + if err != nil { + return 0, fmt.Errorf("ledger: payload: %w", err) + } + payload = string(b) + } + var prevSeq int64 + prev := genesis + err := tx.QueryRowContext(ctx, `SELECT seq, hash FROM ledger ORDER BY seq DESC LIMIT 1`).Scan(&prevSeq, &prev) + if err != nil && !errors.Is(err, sql.ErrNoRows) { + return 0, fmt.Errorf("ledger: read tail: %w", err) + } + seq := prevSeq + 1 + ts := clk.Now().UTC().Format(time.RFC3339) + hash := rowHash(prev, seq, ts, e.Kind, e.OrderID, actor, payload) + var oid any + if e.OrderID != "" { + oid = e.OrderID + } + if _, err := tx.ExecContext(ctx, `INSERT INTO ledger(seq,ts,kind,order_id,actor,payload,prev_hash,hash) VALUES(?,?,?,?,?,?,?,?)`, + seq, ts, e.Kind, oid, actor, payload, prev, hash); err != nil { + return 0, fmt.Errorf("ledger: insert: %w", err) + } + return seq, nil +} + +// Result is the outcome of walking the chain. +type Result struct { + Rows int + OK bool + FirstBadSeq int64 + Reason string +} + +// Verify walks the whole chain and reports the first broken row. +func Verify(ctx context.Context, db *sql.DB) (Result, error) { + rows, err := db.QueryContext(ctx, `SELECT seq,ts,kind,COALESCE(order_id,''),actor,payload,prev_hash,hash FROM ledger ORDER BY seq`) + if err != nil { + return Result{}, err + } + defer rows.Close() + res := Result{OK: true} + prevHash, expectSeq := genesis, int64(1) + for rows.Next() { + var seq int64 + var ts, kind, oid, actor, payload, prev, hash string + if err := rows.Scan(&seq, &ts, &kind, &oid, &actor, &payload, &prev, &hash); err != nil { + return Result{}, err + } + res.Rows++ + bad := func(reason string) (Result, error) { + return Result{Rows: res.Rows, FirstBadSeq: seq, Reason: reason}, nil + } + switch { + case seq != expectSeq: + return bad(fmt.Sprintf("expected row %d but found %d: a row is missing", expectSeq, seq)) + case prev != prevHash: + return bad("previous hash does not match the row before it") + case rowHash(prev, seq, ts, kind, oid, actor, payload) != hash: + return bad("row contents do not match its hash") + } + prevHash, expectSeq = hash, seq+1 + } + return res, rows.Err() +} + +// Describe renders a Result for people. +func (r Result) Describe() string { + if r.OK { + return fmt.Sprintf("ledger intact: %d rows, chain valid", r.Rows) + } + return strings.TrimSpace(fmt.Sprintf("ledger BROKEN at row %d: %s", r.FirstBadSeq, r.Reason)) +} diff --git a/internal/ledger/ledger_test.go b/internal/ledger/ledger_test.go new file mode 100644 index 0000000..498c00d --- /dev/null +++ b/internal/ledger/ledger_test.go @@ -0,0 +1,199 @@ +package ledger + +import ( + "context" + "database/sql" + "errors" + "fmt" + "sync" + "testing" + "time" + + "github.com/opensawce/kaasu/internal/clock" + "github.com/opensawce/kaasu/internal/store" + "github.com/opensawce/kaasu/internal/testutil" +) + +var clk = clock.NewFake(time.Date(2026, 1, 1, 9, 0, 0, 0, time.UTC)) + +func appendN(t *testing.T, s *store.Store, n int) { + t.Helper() + for i := 1; i <= n; i++ { + err := s.Tx(context.Background(), func(tx *sql.Tx) error { + _, err := Append(context.Background(), tx, clk, Entry{Kind: "order.created", OrderID: fmt.Sprintf("ord_%d", i), Payload: map[string]any{"amount_paise": 49900 + i}}) + return err + }) + if err != nil { + t.Fatal(err) + } + clk.Advance(time.Second) + } +} + +func TestAppendBuildsAValidChain(t *testing.T) { + s := testutil.OpenStore(t) + appendN(t, s, 3) + res, err := Verify(context.Background(), s.Read()) + if err != nil || !res.OK || res.Rows != 3 { + t.Fatalf("res=%+v err=%v", res, err) + } + var prev string + _ = s.Read().QueryRow(`SELECT prev_hash FROM ledger WHERE seq=1`).Scan(&prev) + if prev != genesis { + t.Errorf("first prev_hash=%q", prev) + } + empty := testutil.OpenStore(t) + if r, _ := Verify(context.Background(), empty.Read()); !r.OK || r.Rows != 0 { + t.Errorf("empty ledger should verify: %+v", r) + } +} + +func TestTriggersBlockUpdateAndDelete(t *testing.T) { + s := testutil.OpenStore(t) + appendN(t, s, 2) + if _, err := s.Read().Exec(`UPDATE ledger SET kind='x' WHERE seq=1`); err == nil { + t.Error("UPDATE must be blocked") + } + if _, err := s.Read().Exec(`DELETE FROM ledger WHERE seq=1`); err == nil { + t.Error("DELETE must be blocked") + } +} + +// Review focus 4 and issue #9: tamper with any row, verify fails at that row. +func TestTamperingBreaksTheChainAtThatRow(t *testing.T) { + cols := map[string]string{ + "ts": "2030-01-01T00:00:00Z", "kind": "order.confirmed", "order_id": "ord_evil", + "actor": "attacker", "payload": `{"amount_paise":1}`, + } + for row := 1; row <= 4; row++ { + for col, val := range cols { + t.Run(fmt.Sprintf("row%d/%s", row, col), func(t *testing.T) { + s := testutil.OpenStore(t) + appendN(t, s, 4) + mustExec(t, s, `DROP TRIGGER ledger_no_update`) + mustExec(t, s, fmt.Sprintf(`UPDATE ledger SET %s=? WHERE seq=?`, col), val, row) + res, err := Verify(context.Background(), s.Read()) + if err != nil { + t.Fatal(err) + } + if res.OK || res.FirstBadSeq != int64(row) { + t.Errorf("want break at %d, got %+v", row, res) + } + }) + } + } +} + +func TestDeletedRowBreaksTheChain(t *testing.T) { + s := testutil.OpenStore(t) + appendN(t, s, 4) + mustExec(t, s, `DROP TRIGGER ledger_no_delete`) + mustExec(t, s, `DELETE FROM ledger WHERE seq=2`) + res, _ := Verify(context.Background(), s.Read()) + if res.OK || res.FirstBadSeq != 3 { + t.Errorf("want break at 3 (first row after the gap), got %+v", res) + } +} + +func TestRewrittenHashStillBreaksNextRow(t *testing.T) { + s := testutil.OpenStore(t) + appendN(t, s, 3) + mustExec(t, s, `DROP TRIGGER ledger_no_update`) + mustExec(t, s, `UPDATE ledger SET payload='{"amount_paise":1}' WHERE seq=2`) + // A careful attacker also recomputes row 2's hash, but row 3 still points at the old one. + var ts, kind, oid, actor, payload, prev string + _ = s.Read().QueryRow(`SELECT ts,kind,COALESCE(order_id,''),actor,payload,prev_hash FROM ledger WHERE seq=2`).Scan(&ts, &kind, &oid, &actor, &payload, &prev) + mustExec(t, s, `UPDATE ledger SET hash=? WHERE seq=2`, rowHash(prev, 2, ts, kind, oid, actor, payload)) + res, _ := Verify(context.Background(), s.Read()) + if res.OK || res.FirstBadSeq != 3 { + t.Errorf("want break at 3, got %+v", res) + } +} + +func TestAppendRollsBackWithItsTransaction(t *testing.T) { + s := testutil.OpenStore(t) + appendN(t, s, 1) + err := s.Tx(context.Background(), func(tx *sql.Tx) error { + if _, err := Append(context.Background(), tx, clk, Entry{Kind: "order.created", OrderID: "ord_x"}); err != nil { + return err + } + return errors.New("state change failed") + }) + if err == nil { + t.Fatal("expected failure") + } + res, _ := Verify(context.Background(), s.Read()) + if !res.OK || res.Rows != 1 { + t.Errorf("rolled-back append must leave no row: %+v", res) + } +} + +func TestConcurrentAppendsStayChained(t *testing.T) { + s := testutil.OpenStore(t) + var wg sync.WaitGroup + for i := 0; i < 30; i++ { + wg.Add(1) + go func(i int) { + defer wg.Done() + _ = s.Tx(context.Background(), func(tx *sql.Tx) error { + _, err := Append(context.Background(), tx, clk, Entry{Kind: "admin.login", Actor: fmt.Sprintf("admin:%d", i)}) + return err + }) + }(i) + } + wg.Wait() + res, err := Verify(context.Background(), s.Read()) + if err != nil || !res.OK || res.Rows != 30 { + t.Fatalf("res=%+v err=%v", res, err) + } +} + +func TestActorFromContext(t *testing.T) { + s := testutil.OpenStore(t) + ctx := WithActor(context.Background(), "admin:priya") + _ = s.Tx(ctx, func(tx *sql.Tx) error { _, err := Append(ctx, tx, clk, Entry{Kind: "account.created"}); return err }) + _ = s.Tx(context.Background(), func(tx *sql.Tx) error { + _, err := Append(context.Background(), tx, clk, Entry{Kind: "order.expired"}) + return err + }) + var a1, a2 string + _ = s.Read().QueryRow(`SELECT actor FROM ledger WHERE seq=1`).Scan(&a1) + _ = s.Read().QueryRow(`SELECT actor FROM ledger WHERE seq=2`).Scan(&a2) + if a1 != "admin:priya" || a2 != "system" { + t.Errorf("actors: %q %q", a1, a2) + } +} + +func mustExec(t *testing.T, s *store.Store, q string, args ...any) { + t.Helper() + if err := s.Tx(context.Background(), func(tx *sql.Tx) error { _, err := tx.Exec(q, args...); return err }); err != nil { + t.Fatalf("%s: %v", q, err) + } +} + +func TestAppendRejectsTheSeparatorByteAndEmptyKind(t *testing.T) { + s := testutil.OpenStore(t) + for _, e := range []Entry{ + {Kind: "order\x1fcreated"}, + {Kind: "order.created", OrderID: "ord\x1f1"}, + {Kind: "order.created", Actor: "a\x1fb"}, + {}, + } { + err := s.Tx(context.Background(), func(tx *sql.Tx) error { + _, err := Append(context.Background(), tx, clk, e) + return err + }) + if err == nil { + t.Errorf("entry %+v must be rejected", e) + } + } +} + +func TestDescribe(t *testing.T) { + if got := (Result{OK: true, Rows: 3}).Describe(); got != "ledger intact: 3 rows, chain valid" { + t.Errorf("ok: %q", got) + } + if got := (Result{FirstBadSeq: 2, Reason: "row contents do not match its hash"}).Describe(); got != "ledger BROKEN at row 2: row contents do not match its hash" { + t.Errorf("broken: %q", got) + } +} diff --git a/internal/secrets/secrets_test.go b/internal/secrets/secrets_test.go index 7fd4a77..c7fd806 100644 --- a/internal/secrets/secrets_test.go +++ b/internal/secrets/secrets_test.go @@ -5,6 +5,7 @@ import ( "context" "database/sql" "encoding/base64" + "fmt" "os" "path/filepath" "strings" @@ -201,3 +202,30 @@ func TestAPIKeys(t *testing.T) { t.Errorf("revoked key must fail: %v", err) } } + +func TestSecretAndKeyChangesAreLedgeredWithoutValues(t *testing.T) { + v := newVault(t, key(1)) + ctx := context.Background() + _ = v.Put(ctx, "imap_password", []byte("hunter2")) + plain, rec, _ := v.NewAPIKey(ctx, "Woo") + _ = v.RevokeAPIKey(ctx, rec.ID) + _, _ = v.RotateAll(ctx, key(2)) + rows, err := v.st.Read().Query(`SELECT kind, payload FROM ledger ORDER BY seq`) + if err != nil { + t.Fatal(err) + } + defer rows.Close() + var kinds []string + for rows.Next() { + var k, p string + _ = rows.Scan(&k, &p) + kinds = append(kinds, k) + if strings.Contains(p, "hunter2") || strings.Contains(p, plain) { + t.Errorf("ledger payload leaks a secret: %s", p) + } + } + want := "[secret.updated apikey.created apikey.revoked secret.master_key_rotated]" + if fmt.Sprint(kinds) != want { + t.Errorf("kinds=%v want %s", kinds, want) + } +} diff --git a/internal/secrets/vault.go b/internal/secrets/vault.go index e21d803..624c1cc 100644 --- a/internal/secrets/vault.go +++ b/internal/secrets/vault.go @@ -15,6 +15,7 @@ import ( "github.com/opensawce/kaasu/internal/clock" "github.com/opensawce/kaasu/internal/id" + "github.com/opensawce/kaasu/internal/ledger" "github.com/opensawce/kaasu/internal/store" ) @@ -52,7 +53,10 @@ func (v *Vault) Put(ctx context.Context, name string, plaintext []byte) error { return v.st.Tx(ctx, func(tx *sql.Tx) error { _, err := tx.ExecContext(ctx, `INSERT INTO secrets(name,blob,updated_at) VALUES(?,?,?) ON CONFLICT(name) DO UPDATE SET blob=excluded.blob, updated_at=excluded.updated_at`, name, blob, v.now()) - return err + if err != nil { + return err + } + return v.record(ctx, tx, "secret.updated", map[string]any{"name": name}) }) } @@ -105,7 +109,7 @@ func (v *Vault) RotateAll(ctx context.Context, next MasterKey) (int, error) { } } count = len(all) - return nil + return v.record(ctx, tx, "secret.master_key_rotated", map[string]any{"secrets": count}) }) if err != nil { return 0, err @@ -139,7 +143,10 @@ func (v *Vault) NewAPIKey(ctx context.Context, label string) (string, APIKey, er err := v.st.Tx(ctx, func(tx *sql.Tx) error { _, err := tx.ExecContext(ctx, `INSERT INTO api_keys(id,label,prefix,key_hash,created_at) VALUES(?,?,?,?,?)`, rec.ID, rec.Label, rec.Prefix, hashKey(plain), v.now()) - return err + if err != nil { + return err + } + return v.record(ctx, tx, "apikey.created", map[string]any{"key_id": rec.ID, "label": label, "prefix": rec.Prefix}) }) if err != nil { return "", APIKey{}, err @@ -170,6 +177,12 @@ func (v *Vault) RevokeAPIKey(ctx context.Context, keyID string) error { if n, _ := res.RowsAffected(); n == 0 { return ErrInvalidKey } - return nil + return v.record(ctx, tx, "apikey.revoked", map[string]any{"key_id": keyID}) }) } + +// record ledgers an admin action. Payloads carry names and ids, never secret values. +func (v *Vault) record(ctx context.Context, tx *sql.Tx, kind string, payload map[string]any) error { + _, err := ledger.Append(ctx, tx, v.clk, ledger.Entry{Kind: kind, Payload: payload}) + return err +} diff --git a/internal/store/migrations/0003_ledger_append_only.sql b/internal/store/migrations/0003_ledger_append_only.sql new file mode 100644 index 0000000..acd145f --- /dev/null +++ b/internal/store/migrations/0003_ledger_append_only.sql @@ -0,0 +1,7 @@ +-- The ledger is append-only. A person with the file can drop these triggers, which is why +-- every row is also hash-chained and `kaasu verify` checks the chain. +CREATE TRIGGER ledger_no_update BEFORE UPDATE ON ledger +BEGIN SELECT RAISE(ABORT, 'ledger is append-only'); END; + +CREATE TRIGGER ledger_no_delete BEFORE DELETE ON ledger +BEGIN SELECT RAISE(ABORT, 'ledger is append-only'); END; From a4f23aaf9c7058dd8b0bcda1afdd072c323c2761 Mon Sep 17 00:00:00 2001 From: Manish <69393822+manish-9245@users.noreply.github.com> Date: Sat, 24 Oct 2026 12:24:57 +0530 Subject: [PATCH 09/54] feat(api): orders API with idempotency (#94) * feat(api): orders API with idempotency, UPI links and API-key auth * fix: surface openApp errors, bound limit and expiry inputs, log 500s, context-aware test calls --- cmd/kaasu/main.go | 105 ++++++++++++-- cmd/kaasu/main_test.go | 20 +++ internal/api/api.go | 191 +++++++++++++++++++++++++ internal/api/api_test.go | 162 +++++++++++++++++++++ internal/app/app.go | 62 +++++++++ internal/app/app_test.go | 69 +++++++++ internal/orders/orders.go | 247 +++++++++++++++++++++++++++++++++ internal/orders/orders_test.go | 157 +++++++++++++++++++++ internal/server/server.go | 32 ++++- internal/server/server_test.go | 19 +++ internal/upi/upi.go | 29 ++++ internal/upi/upi_test.go | 61 ++++++++ openapi.yaml | 89 ++++++++++++ 13 files changed, 1224 insertions(+), 19 deletions(-) create mode 100644 internal/api/api.go create mode 100644 internal/api/api_test.go create mode 100644 internal/app/app.go create mode 100644 internal/app/app_test.go create mode 100644 internal/orders/orders.go create mode 100644 internal/orders/orders_test.go create mode 100644 internal/upi/upi.go create mode 100644 internal/upi/upi_test.go create mode 100644 openapi.yaml diff --git a/cmd/kaasu/main.go b/cmd/kaasu/main.go index db8b1c2..7f20d5d 100644 --- a/cmd/kaasu/main.go +++ b/cmd/kaasu/main.go @@ -5,26 +5,29 @@ import ( "context" "crypto/rand" "encoding/base64" + "flag" "fmt" "io" "log/slog" + "math" "os" "os/signal" "syscall" "time" + "github.com/opensawce/kaasu/internal/accounts" + "github.com/opensawce/kaasu/internal/app" "github.com/opensawce/kaasu/internal/clock" "github.com/opensawce/kaasu/internal/config" "github.com/opensawce/kaasu/internal/ledger" "github.com/opensawce/kaasu/internal/secrets" - "github.com/opensawce/kaasu/internal/server" "github.com/opensawce/kaasu/internal/store" ) // version is set at build time with -ldflags "-X main.version=...". var version = "dev" -const usage = "usage: kaasu [flags]" +const usage = "usage: kaasu [flags]" func main() { os.Exit(run(os.Args[1:], os.Getenv, os.Stdout, os.Stderr)) @@ -53,6 +56,10 @@ func run(args []string, getenv func(string) string, stdout, stderr io.Writer) in return 0 case "rotate-key": return rotateKey(getenv, stdout, stderr) + case "account": + return account(args[1:], getenv, stdout, stderr) + case "apikey": + return apikey(args[1:], getenv, stdout, stderr) case "verify": return verify(getenv, stdout, stderr) case "backup", "pair": @@ -146,22 +153,94 @@ func verify(getenv func(string) string, stdout, stderr io.Writer) int { return 0 } -func serve(args []string, getenv func(string) string, stderr io.Writer) int { - if len(args) > 0 { - fmt.Fprintf(stderr, "serve: unexpected arguments %q (configure with KAASU_CONFIG or KAASU_* env)\n", args) - return 2 - } +// openApp loads config and the master key, then builds the app (no listeners yet). +// openApp reports failures on stderr; quiet also silences the startup log lines. +func openApp(getenv func(string) string, stderr io.Writer, quiet bool) (*app.App, *slog.Logger, bool) { cfg, err := config.Load(getenv, getenv("KAASU_CONFIG")) if err != nil { fmt.Fprintln(stderr, "config:", err) - return 1 + return nil, nil, false + } + key, source, err := secrets.LoadMasterKey(getenv, secrets.MacKeychain) + if err != nil { + fmt.Fprintln(stderr, "master key:", err) + return nil, nil, false } var level slog.Level _ = level.UnmarshalText([]byte(cfg.LogLevel)) - log := slog.New(slog.NewJSONHandler(stderr, &slog.HandlerOptions{Level: level})) + var logw io.Writer = stderr + if quiet { + logw = io.Discard + } + log := slog.New(slog.NewJSONHandler(logw, &slog.HandlerOptions{Level: level})) + log.Info("master key loaded", "source", source) + a, err := app.New(cfg, key, log) + if err != nil { + fmt.Fprintln(stderr, err) + return nil, nil, false + } + return a, log, true +} - s := server.New(cfg, log) - if err := s.Start(); err != nil { +func account(args []string, getenv func(string) string, stdout, stderr io.Writer) int { + if len(args) == 0 || args[0] != "add" { + fmt.Fprintln(stderr, "usage: kaasu account add --vpa --bank --name [--limit-rupees N]") + return 2 + } + fs := flag.NewFlagSet("account add", flag.ContinueOnError) + fs.SetOutput(stderr) + vpa, bank, name := fs.String("vpa", "", "UPI ID"), fs.String("bank", "", "bank name"), fs.String("name", "", "display name") + limit := fs.Int64("limit-rupees", 0, "orders above this need your OK (0 = always ask)") + if err := fs.Parse(args[1:]); err != nil { + return 2 + } + a, _, ok := openApp(getenv, stderr, true) + if !ok { + return 1 + } + defer a.Store.Close() + if *limit < 0 || *limit > math.MaxInt64/100 { + fmt.Fprintln(stderr, "account add: --limit-rupees must be between 0 and", int64(math.MaxInt64/100)) + return 2 + } + acc, err := a.Accounts.Create(ledger.WithActor(context.Background(), "cli"), accounts.Input{VPA: *vpa, Bank: *bank, DisplayName: *name, AutoConfirmLimitPaise: *limit * 100}) + if err != nil { + fmt.Fprintln(stderr, "account add:", err) + return 1 + } + fmt.Fprintf(stdout, "added %s (%s)\n", acc.ID, acc.VPA) + return 0 +} + +func apikey(args []string, getenv func(string) string, stdout, stderr io.Writer) int { + if len(args) != 2 || args[0] != "new" { + fmt.Fprintln(stderr, "usage: kaasu apikey new