// Package authd — sources.go: Source CRUD for M13b W2. // // Sources are the *runtime* ingest-side counterpart of tenants. // Each (company_id, id) row represents a single source that can // POST events into ingestd. M13b W2 turns the existing // public.sources table (created in migration 003) into a // fully-managed resource in the admin UI. // // Schema (post-migration 011): // id TEXT // company_id TEXT FK -> public.companies(id) // name TEXT // type TEXT (http | mqtt | ws | grpc) // rate_limit_per_sec INTEGER // allowed_targets JSONB (M13c routing UI will edit; W2 read-only) // match_expr JSONB (M13c routing UI will edit; W2 read-only) // status TEXT (active | suspended) // hmac_secret_hash TEXT (bcrypt; returned only on create/rotate) // api_key_hash TEXT (bcrypt; returned only on create/rotate) // mtls_required BOOLEAN (M14 reads; W2 just stores) // description TEXT // created_at TIMESTAMPTZ // // Bridge to auth.tenants: // auth.tenants.id is a UUID; public.sources.company_id is a // TEXT FK to public.companies.id (also TEXT). The two schemas // predate each other and were never formally linked. The // convention this code enforces: public.companies.id == // auth.tenants.id::text. CreateSource uses ON CONFLICT // DO NOTHING to ensure a public.companies row exists before // the FK is hit, so creating a source for an auth tenant // without a corresponding public.companies row is a no-op // (idempotent) rather than an error. // // Threading: safe for concurrent use (pgx pool is goroutine-safe). package authd import ( "context" "crypto/rand" "encoding/hex" "encoding/json" "golang.org/x/crypto/bcrypt" "errors" "fmt" "strings" "time" "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgconn" ) // Source is the wire shape returned to handlers / JSON callers. // Secrets are NEVER included — only the booleans flagging their // presence (`hmac_set`, `api_key_set`). Plaintext values live // in the one-time SecretsPayload returned by CreateSource and // RotateSecrets. type Source struct { ID string `json:"id"` CompanyID string `json:"company_id"` Name string `json:"name"` Type string `json:"type"` RateLimitPerSec int `json:"rate_limit_per_sec"` AllowedTargets json.RawMessage `json:"allowed_targets"` MatchExpr json.RawMessage `json:"match_expr"` Status string `json:"status"` MTLSRequired bool `json:"mtls_required"` Description string `json:"description,omitempty"` HMACSet bool `json:"hmac_set"` APIKeySet bool `json:"api_key_set"` CreatedAt time.Time `json:"created_at"` } // SecretsPayload is the one-time plaintext payload returned at // create and rotate time. The UI shows it in a modal that // requires the operator to confirm "I have saved these before // continuing." After that, the values are gone from the server // and the modal can never re-display them. type SecretsPayload struct { HMACSecret string `json:"hmac_secret"` APIKey string `json:"api_key"` } // ErrSourceNotFound is returned when (company_id, id) doesn't exist. var ErrSourceNotFound = errors.New("authd: source not found") // ErrSourceIDTaken is returned when CreateSource sees a // duplicate (company_id, id) for a tenant that already has it. var ErrSourceIDTaken = errors.New("authd: source id already in use") // ErrSourceInvalid is returned when input validation fails. var ErrSourceInvalid = errors.New("authd: source input invalid") // validSourceTypes is the whitelist of `type` values. Mirrors // the schema default comment in 003. var validSourceTypes = map[string]struct{}{ "http": {}, "mqtt": {}, "ws": {}, "grpc": {}, } // SourceFilter controls ListSources. Empty fields mean "no filter". type SourceFilter struct { Q string // matches id OR name (ILIKE) Type string // exact match Status string // exact match Limit int Offset int CallerRole string CallerTenant string // auth.tenants.id (UUID string) } // CreateSourceInput is the validated create payload. The optional // HMAC + API key fields, if non-empty, are bcrypt-hashed by // CreateSource. The plaintext is NEVER persisted; the caller // receives it in the returned SecretsPayload. type CreateSourceInput struct { ID string Name string Type string RateLimitPerSec int AllowedTargets json.RawMessage MatchExpr json.RawMessage Description string MTLSRequired bool HMACSecret string // optional; if non-empty, will be hashed APIKey string // optional; if non-empty, will be hashed } // UpdateSourceInput is the PATCH payload. Pointer / non-nil // fields mean "apply this." Nil raw-message means "leave the // JSON column as-is." All fields are optional; an empty patch // is a no-op. type UpdateSourceInput struct { Name *string Type *string RateLimitPerSec *int Description *string MTLSRequired *bool AllowedTargets json.RawMessage MatchExpr json.RawMessage } // Validate runs the constraints the DB enforces, but earlier // and with friendlier error messages for the UI. func (in *CreateSourceInput) Validate() error { if !validSourceID(in.ID) { return fmt.Errorf("%w: id must match ^[a-z0-9][a-z0-9-]{0,62}[a-z0-9]$", ErrSourceInvalid) } if strings.TrimSpace(in.Name) == "" { return fmt.Errorf("%w: name is required", ErrSourceInvalid) } if len(in.Name) > 200 { return fmt.Errorf("%w: name must be \u2264 200 characters", ErrSourceInvalid) } if _, ok := validSourceTypes[in.Type]; !ok { return fmt.Errorf("%w: type must be http|mqtt|ws|grpc", ErrSourceInvalid) } if in.RateLimitPerSec < 1 || in.RateLimitPerSec > 1_000_000 { return fmt.Errorf("%w: rate_limit_per_sec must be 1..1000000", ErrSourceInvalid) } if len(in.Description) > 500 { return fmt.Errorf("%w: description must be \u2264 500 characters", ErrSourceInvalid) } // Empty raw messages are allowed; CreateSource defaults them to '[]' / '{}'. if len(in.AllowedTargets) > 0 && !json.Valid(in.AllowedTargets) { return fmt.Errorf("%w: allowed_targets must be valid JSON", ErrSourceInvalid) } if len(in.MatchExpr) > 0 && !json.Valid(in.MatchExpr) { return fmt.Errorf("%w: match_expr must be valid JSON", ErrSourceInvalid) } if in.HMACSecret != "" && !validSecretFormat(in.HMACSecret) { return fmt.Errorf("%w: hmac_secret, if provided, must be 32..128 [A-Za-z0-9_-] chars", ErrSourceInvalid) } if in.APIKey != "" && !validAPIKeyFormat(in.APIKey) { return fmt.Errorf("%w: api_key, if provided, must be 16..128 [A-Za-z0-9_-] chars", ErrSourceInvalid) } return nil } // Validate is the same for Update. We don't enforce presence // of fields (PATCH can be empty), just per-field constraints. func (in *UpdateSourceInput) Validate() error { if in.Name != nil { s := strings.TrimSpace(*in.Name) if s == "" { return fmt.Errorf("%w: name cannot be empty", ErrSourceInvalid) } if len(s) > 200 { return fmt.Errorf("%w: name must be \u2264 200 characters", ErrSourceInvalid) } } if in.Type != nil { if _, ok := validSourceTypes[*in.Type]; !ok { return fmt.Errorf("%w: type must be http|mqtt|ws|grpc", ErrSourceInvalid) } } if in.RateLimitPerSec != nil && (*in.RateLimitPerSec < 1 || *in.RateLimitPerSec > 1_000_000) { return fmt.Errorf("%w: rate_limit_per_sec must be 1..1000000", ErrSourceInvalid) } if in.Description != nil && len(*in.Description) > 500 { return fmt.Errorf("%w: description must be \u2264 500 characters", ErrSourceInvalid) } if in.AllowedTargets != nil && !json.Valid(in.AllowedTargets) { return fmt.Errorf("%w: allowed_targets must be valid JSON", ErrSourceInvalid) } if in.MatchExpr != nil && !json.Valid(in.MatchExpr) { return fmt.Errorf("%w: match_expr must be valid JSON", ErrSourceInvalid) } return nil } // ensurePublicCompanyRow makes sure public.companies has a row // keyed by the auth.tenants.id (cast to text). This is the // bridge between the M13a auth schema and the M0 data-plane // schema. Idempotent; safe to call from CreateSource. The row // carries just the minimum data: id, name (display_name), a // default rate_limit, status='active'. Operators who want to // manage the public.companies row's fields (rate_limit, etc.) // can do so via the W1 tenant API; this code path only ensures // the FK target exists. func (s *Store) ensurePublicCompanyRow(ctx context.Context, tenantID, displayName string) error { if s.pool == nil { return errors.New("authd: no DB pool (test mode)") } const q = ` INSERT INTO public.companies (id, name, status, rate_limit_per_sec) VALUES ($1::text, $2, 'active', 10000) ON CONFLICT (id) DO NOTHING ` _, err := s.pool.Exec(ctx, q, tenantID, displayName) if err != nil { return fmt.Errorf("ensure public.companies row: %w", err) } return nil } // ListSources returns sources visible to the caller under the // given filter, plus the total count. Scope: super_admin sees // all tenants' sources; non-super_admin sees only the caller's // tenant. CallerTenant is the auth.tenants.id (UUID string). func (s *Store) ListSources(ctx context.Context, f SourceFilter) ([]Source, int, error) { if s.pool == nil { return nil, 0, errors.New("authd: no DB pool (test mode)") } if f.Limit <= 0 { f.Limit = 100 } if f.Limit > 500 { f.Limit = 500 } args := []any{} conds := []string{} if f.CallerRole != "super_admin" { if f.CallerTenant == "" { return []Source{}, 0, nil } // Bridge: public.sources.company_id (TEXT) = auth.tenants.id::text args = append(args, f.CallerTenant) conds = append(conds, fmt.Sprintf("company_id = $%d", len(args))) } if strings.TrimSpace(f.Type) != "" { args = append(args, f.Type) conds = append(conds, fmt.Sprintf("type = $%d", len(args))) } if strings.TrimSpace(f.Status) != "" { args = append(args, f.Status) conds = append(conds, fmt.Sprintf("status = $%d", len(args))) } if strings.TrimSpace(f.Q) != "" { args = append(args, "%"+strings.TrimSpace(f.Q)+"%") conds = append(conds, fmt.Sprintf("(id ILIKE $%d OR name ILIKE $%d)", len(args), len(args))) } where := "" if len(conds) > 0 { where = "WHERE " + strings.Join(conds, " AND ") } var total int if err := s.pool.QueryRow(ctx, "SELECT COUNT(*) FROM public.sources "+where, args...).Scan(&total); err != nil { return nil, 0, fmt.Errorf("count sources: %w", err) } args = append(args, f.Limit, f.Offset) q := fmt.Sprintf(` SELECT id, company_id, name, type, rate_limit_per_sec, allowed_targets, match_expr, status, mtls_required, COALESCE(description, ''), (hmac_secret_hash IS NOT NULL), (api_key_hash IS NOT NULL), created_at FROM public.sources %s ORDER BY created_at DESC LIMIT $%d OFFSET $%d `, where, len(args)-1, len(args)) rows, err := s.pool.Query(ctx, q, args...) if err != nil { return nil, 0, fmt.Errorf("list sources: %w", err) } defer rows.Close() out := make([]Source, 0, f.Limit) for rows.Next() { var src Source if err := rows.Scan( &src.ID, &src.CompanyID, &src.Name, &src.Type, &src.RateLimitPerSec, &src.AllowedTargets, &src.MatchExpr, &src.Status, &src.MTLSRequired, &src.Description, &src.HMACSet, &src.APIKeySet, &src.CreatedAt, ); err != nil { return nil, 0, fmt.Errorf("scan source: %w", err) } out = append(out, src) } if err := rows.Err(); err != nil { return nil, 0, fmt.Errorf("rows: %w", err) } return out, total, nil } // GetSource fetches a single source by (company_id, id). Returns // ErrSourceNotFound if missing. Caller is responsible for the // per-id scope check (canAccessSource); this method is a // straight DB lookup. func (s *Store) GetSource(ctx context.Context, companyID, id string) (*Source, error) { if s.pool == nil { return nil, errors.New("authd: no DB pool (test mode)") } const q = ` SELECT id, company_id, name, type, rate_limit_per_sec, allowed_targets, match_expr, status, mtls_required, COALESCE(description, ''), (hmac_secret_hash IS NOT NULL), (api_key_hash IS NOT NULL), created_at FROM public.sources WHERE company_id = $1 AND id = $2 ` src := &Source{} err := s.pool.QueryRow(ctx, q, companyID, id).Scan( &src.ID, &src.CompanyID, &src.Name, &src.Type, &src.RateLimitPerSec, &src.AllowedTargets, &src.MatchExpr, &src.Status, &src.MTLSRequired, &src.Description, &src.HMACSet, &src.APIKeySet, &src.CreatedAt, ) if err != nil { if errors.Is(err, pgx.ErrNoRows) { return nil, ErrSourceNotFound } return nil, fmt.Errorf("get source: %w", err) } return src, nil } // CreateSource inserts a new source and returns the row plus a // one-time SecretsPayload. The bridge to public.companies is // handled inside (ensurePublicCompanyRow). Audit log written. // // Behavior: // - Duplicate (company_id, id) returns ErrSourceIDTaken (409). // - If HMACSecret / APIKey are empty in the input, no hash is // stored; the resulting Source has hmac_set=false. // - If non-empty, they're bcrypt-hashed at cost 10 and the // plaintext is returned in SecretsPayload. The plaintext // is the ONLY time the UI can see it. func (s *Store) CreateSource( ctx context.Context, tenantID, tenantDisplayName string, in CreateSourceInput, actorUserID, actorIP, actorUA string, ) (*Source, *SecretsPayload, error) { if s.pool == nil { return nil, nil, errors.New("authd: no DB pool (test mode)") } if err := in.Validate(); err != nil { return nil, nil, err } // Bridge: ensure public.companies has a row keyed by the // auth.tenants.id cast to text. This is the only place // authd writes to public.companies; everything else is // via the M13a auth.tenants API. if err := s.ensurePublicCompanyRow(ctx, tenantID, tenantDisplayName); err != nil { return nil, nil, err } // Default the JSONB columns if the caller didn't send // anything: '[]' for allowed_targets, '{}' for match_expr. allowedTargets := in.AllowedTargets if len(allowedTargets) == 0 { allowedTargets = json.RawMessage(`[]`) } matchExpr := in.MatchExpr if len(matchExpr) == 0 { matchExpr = json.RawMessage(`{}`) } // Hash the optional secrets. cost=10 mirrors the bootstrap // path; v1.1 will bump to 12 in prod. hmacHash, err := hashSecret(in.HMACSecret, "hmac") if err != nil { return nil, nil, err } apiKeyHash, err := hashSecret(in.APIKey, "api_key") if err != nil { return nil, nil, err } const q = ` INSERT INTO public.sources (company_id, id, name, type, rate_limit_per_sec, allowed_targets, match_expr, status, mtls_required, description, hmac_secret_hash, api_key_hash) VALUES ($1::text, $2, $3, $4, $5, $6, $7, 'active', $8, NULLIF($9, ''), $10, $11) RETURNING id, company_id, name, type, rate_limit_per_sec, allowed_targets, match_expr, status, mtls_required, COALESCE(description, ''), (hmac_secret_hash IS NOT NULL), (api_key_hash IS NOT NULL), created_at ` src := &Source{} err = s.pool.QueryRow(ctx, q, tenantID, in.ID, in.Name, in.Type, in.RateLimitPerSec, allowedTargets, matchExpr, in.MTLSRequired, in.Description, nullableString(hmacHash), nullableString(apiKeyHash), ).Scan( &src.ID, &src.CompanyID, &src.Name, &src.Type, &src.RateLimitPerSec, &src.AllowedTargets, &src.MatchExpr, &src.Status, &src.MTLSRequired, &src.Description, &src.HMACSet, &src.APIKeySet, &src.CreatedAt, ) if err != nil { var pgErr *pgconn.PgError if errors.As(err, &pgErr) && pgErr.Code == "23505" { return nil, nil, ErrSourceIDTaken } return nil, nil, fmt.Errorf("create source: %w", err) } // Audit. The plaintext secrets are NOT included in the // audit payload — we only audit the fact that a source // was created, not the values. if err := s.WriteAudit(ctx, "source.create", actorUserID, actorIP, actorUA, src.CompanyID, src.ID, map[string]any{ "name": src.Name, "type": src.Type, "rate_limit_per_sec": src.RateLimitPerSec, "mtls_required": src.MTLSRequired, "hmac_set": src.HMACSet, "api_key_set": src.APIKeySet, }); err != nil { _ = err } // Build the secrets payload only if the caller actually // provided one. Empty input => no payload, so the UI knows // not to render the secrets modal. var payload *SecretsPayload if in.HMACSecret != "" || in.APIKey != "" { payload = &SecretsPayload{ HMACSecret: in.HMACSecret, APIKey: in.APIKey, } } return src, payload, nil } // UpdateSource applies a partial update and writes an audit row. func (s *Store) UpdateSource( ctx context.Context, companyID, id string, in UpdateSourceInput, actorUserID, actorIP, actorUA string, ) (*Source, error) { if s.pool == nil { return nil, errors.New("authd: no DB pool (test mode)") } if err := in.Validate(); err != nil { return nil, err } sets := []string{} args := []any{companyID, id} if in.Name != nil { args = append(args, strings.TrimSpace(*in.Name)) sets = append(sets, fmt.Sprintf("name = $%d", len(args))) } if in.Type != nil { args = append(args, *in.Type) sets = append(sets, fmt.Sprintf("type = $%d", len(args))) } if in.RateLimitPerSec != nil { args = append(args, *in.RateLimitPerSec) sets = append(sets, fmt.Sprintf("rate_limit_per_sec = $%d", len(args))) } if in.Description != nil { args = append(args, *in.Description) sets = append(sets, fmt.Sprintf("description = $%d", len(args))) } if in.MTLSRequired != nil { args = append(args, *in.MTLSRequired) sets = append(sets, fmt.Sprintf("mtls_required = $%d", len(args))) } if in.AllowedTargets != nil { args = append(args, in.AllowedTargets) sets = append(sets, fmt.Sprintf("allowed_targets = $%d", len(args))) } if in.MatchExpr != nil { args = append(args, in.MatchExpr) sets = append(sets, fmt.Sprintf("match_expr = $%d", len(args))) } if len(sets) == 0 { return s.GetSource(ctx, companyID, id) } q := fmt.Sprintf("UPDATE public.sources SET %s WHERE company_id = $1 AND id = $2", strings.Join(sets, ", ")) tag, err := s.pool.Exec(ctx, q, args...) if err != nil { return nil, fmt.Errorf("update source: %w", err) } if tag.RowsAffected() == 0 { return nil, ErrSourceNotFound } payload := map[string]any{} if in.Name != nil { payload["name"] = *in.Name } if in.Type != nil { payload["type"] = *in.Type } if in.RateLimitPerSec != nil { payload["rate_limit_per_sec"] = *in.RateLimitPerSec } if in.Description != nil { payload["description"] = *in.Description } if in.MTLSRequired != nil { payload["mtls_required"] = *in.MTLSRequired } if in.AllowedTargets != nil { payload["allowed_targets_set"] = true } if in.MatchExpr != nil { payload["match_expr_set"] = true } if err := s.WriteAudit(ctx, "source.update", actorUserID, actorIP, actorUA, companyID, id, payload); err != nil { _ = err } return s.GetSource(ctx, companyID, id) } // SetSourceStatus flips status. Allowed transitions: // active -> suspended // suspended -> active // No archive state for sources (per the schema; only // active|suspended). func (s *Store) SetSourceStatus( ctx context.Context, companyID, id, newStatus string, actorUserID, actorIP, actorUA string, ) (*Source, error) { if s.pool == nil { return nil, errors.New("authd: no DB pool (test mode)") } switch newStatus { case "active", "suspended": default: return nil, fmt.Errorf("%w: status must be active|suspended", ErrSourceInvalid) } cur, err := s.GetSource(ctx, companyID, id) if err != nil { return nil, err } if cur.Status == newStatus { return cur, nil } if _, err := s.pool.Exec(ctx, "UPDATE public.sources SET status = $3 WHERE company_id = $1 AND id = $2", companyID, id, newStatus); err != nil { return nil, fmt.Errorf("set source status: %w", err) } if err := s.WriteAudit(ctx, "source.status", actorUserID, actorIP, actorUA, companyID, id, map[string]any{ "from": cur.Status, "to": newStatus, }); err != nil { _ = err } return s.GetSource(ctx, companyID, id) } // RotateSecrets generates a new HMAC secret + API key, hashes // them, and returns the plaintext ONCE in SecretsPayload. The // old secrets are immediately invalidated (overwritten in the // DB). Use this when a source's credentials are suspected to // have leaked. func (s *Store) RotateSecrets( ctx context.Context, companyID, id string, actorUserID, actorIP, actorUA string, ) (*Source, *SecretsPayload, error) { if s.pool == nil { return nil, nil, errors.New("authd: no DB pool (test mode)") } // Confirm the source exists before we generate anything. // If it doesn't, we don't want to surface that a random // pair was generated and then discarded. cur, err := s.GetSource(ctx, companyID, id) if err != nil { return nil, nil, err } hmacPlain, err := generateSecret(32) if err != nil { return nil, nil, fmt.Errorf("generate hmac: %w", err) } apiPlain, err := generateAPIKey() if err != nil { return nil, nil, fmt.Errorf("generate api_key: %w", err) } hmacHash, err := hashSecret(hmacPlain, "hmac") if err != nil { return nil, nil, err } apiKeyHash, err := hashSecret(apiPlain, "api_key") if err != nil { return nil, nil, err } if _, err := s.pool.Exec(ctx, "UPDATE public.sources SET hmac_secret_hash = $3, api_key_hash = $4 WHERE company_id = $1 AND id = $2", companyID, id, hmacHash, apiKeyHash); err != nil { return nil, nil, fmt.Errorf("rotate secrets: %w", err) } if err := s.WriteAudit(ctx, "source.rotate_secrets", actorUserID, actorIP, actorUA, companyID, id, map[string]any{ "hmac_rotated": true, "api_key_rotated": true, }); err != nil { _ = err } updated, err := s.GetSource(ctx, companyID, id) if err != nil { return nil, nil, err } _ = cur return updated, &SecretsPayload{ HMACSecret: hmacPlain, APIKey: apiPlain, }, nil } // ------------------------------------------------------------------- // helpers // ------------------------------------------------------------------- // validSourceID matches the same regex as auth.tenants.slug: // ^[a-z0-9][a-z0-9-]{0,62}[a-z0-9]$ func validSourceID(s string) bool { if len(s) < 2 || len(s) > 64 { return false } if !isAlnumOrDash(s[0]) || s[0] == '-' { return false } if !isAlnumOrDash(s[len(s)-1]) || s[len(s)-1] == '-' { return false } for i := 1; i < len(s)-1; i++ { if !isAlnumOrDash(s[i]) { return false } } return true } // validSecretFormat — HMAC secret: 32..128 [A-Za-z0-9_-] chars. // Long enough to be cryptographically meaningful, short enough // to paste into a config file by hand. func validSecretFormat(s string) bool { if len(s) < 32 || len(s) > 128 { return false } for i := 0; i < len(s); i++ { c := s[i] if !((c >= 'a' && c <= 'z') || (c >= 'A' && c <= 'Z') || (c >= '0' && c <= '9') || c == '_' || c == '-') { return false } } return true } // validAPIKeyFormat — API key: 16..128 [A-Za-z0-9_-] chars. func validAPIKeyFormat(s string) bool { if len(s) < 16 || len(s) > 128 { return false } for i := 0; i < len(s); i++ { c := s[i] if !((c >= 'a' && c <= 'z') || (c >= 'A' && c <= 'Z') || (c >= '0' && c <= '9') || c == '_' || c == '-') { return false } } return true } // generateSecret returns n random bytes hex-encoded. func generateSecret(n int) (string, error) { buf := make([]byte, n) if _, err := rand.Read(buf); err != nil { return "", err } return hex.EncodeToString(buf), nil } // generateAPIKey returns 24 random bytes hex-encoded (48 chars). // Hex is fine because the key never appears in URLs. func generateAPIKey() (string, error) { return generateSecret(24) } // hashSecret bcrypts the plaintext at cost 10. Returns the // empty string if the plaintext is empty (caller checks // separately for "no secret was provided"). The `kind` arg is // only used in the error path; the actual hash doesn't care. func hashSecret(plain, kind string) (string, error) { if plain == "" { return "", nil } h, err := bcrypt.GenerateFromPassword([]byte(plain), 10) if err != nil { return "", fmt.Errorf("hash %s secret: %w", kind, err) } return string(h), nil } // nullableString returns nil for empty input, otherwise &s. // Used to pass optional columns to pgx Exec / QueryRow. func nullableString(s string) any { if s == "" { return nil } return s }