Phase 2: secure OIDC auth and PostgreSQL task persistence

.env.example

@@ -7,3 +7,19 @@ A2A_KEEP_ALIVE_INTERVAL=15s
 A2A_AGENT_INACTIVITY_TIMEOUT=5m
 HTTP_READ_TIMEOUT=30s
 MAX_REQUEST_BODY_BYTES=1048576
+
+OIDC_ISSUER=
+OIDC_AUDIENCE=
+OIDC_TENANT_CLAIM=tenant_id
+OIDC_REQUIRED_SCOPES=a2a
+OIDC_ALLOWED_ALGORITHMS=RS256
+OIDC_CLOCK_SKEW=1m
+OIDC_HTTP_TIMEOUT=10s
+
+DATABASE_URL=
+DATABASE_PASSWORD_FILE=
+DATABASE_MAX_CONNECTIONS=20
+DATABASE_MIN_CONNECTIONS=2
+DATABASE_MAX_CONNECTION_LIFETIME=1h
+DATABASE_MAX_CONNECTION_IDLE=15m
+DATABASE_HEALTH_CHECK_PERIOD=30s

README.md

@@ -26,3 +26,20 @@ for the SDK boundary and planned production adapters.
 
 `.env.example` documents variables loaded by the container runtime. When
 running the binary directly, export overrides in the shell environment.
+
+## OIDC and PostgreSQL development wiring
+
+Set `OIDC_ISSUER`, `OIDC_AUDIENCE`, and `DATABASE_URL` to exercise the durable,
+authenticated path. Credentials are read from mounted files such as
+`DATABASE_PASSWORD_FILE`; passwords embedded in `DATABASE_URL` are rejected.
+Apply schema migrations separately before starting the server:
+
+```powershell
+go run ./cmd/migrate
+go run ./cmd/server
+```
+
+The local in-memory task store remains available only when no database is
+configured in development or tests. Staging and production require explicit
+authentication and durable storage. See [docs/architecture.md](docs/architecture.md)
+for the token and tenant-isolation contract.

cmd/migrate/main.go

@@ -0,0 +1,47 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"log/slog"
+	"os"
+	"os/signal"
+	"strings"
+	"syscall"
+
+	"github.com/KB01111/A2A-RedPandaServer-Container/internal/storage/postgres"
+)
+
+func main() {
+	logger := slog.New(slog.NewJSONHandler(os.Stdout, nil))
+	if err := run(); err != nil {
+		logger.Error("database migration failed", "error", err)
+		os.Exit(1)
+	}
+	logger.Info("database schema is current", "version", postgres.CurrentSchemaVersion())
+}
+
+func run() error {
+	ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
+	defer stop()
+
+	databaseURL := strings.TrimSpace(os.Getenv("DATABASE_URL"))
+	if databaseURL == "" {
+		return fmt.Errorf("DATABASE_URL is required")
+	}
+	pool, err := postgres.OpenPool(ctx, postgres.PoolConfig{
+		DatabaseURL:     databaseURL,
+		PasswordFile:    strings.TrimSpace(os.Getenv("DATABASE_PASSWORD_FILE")),
+		ApplicationName: "bridge-a2a-migrate",
+		MaxConns:        1,
+	})
+	if err != nil {
+		return err
+	}
+	defer pool.Close()
+
+	if err := postgres.Migrate(ctx, pool); err != nil {
+		return err
+	}
+	return postgres.VerifySchema(ctx, pool)
+}

cmd/server/main.go

@@ -3,37 +3,86 @@ package main
 import (
 	"context"
 	"errors"
+	"fmt"
 	"log/slog"
 	"net/http"
 	"os"
 	"os/signal"
 	"syscall"
 	"time"
 
+	"github.com/KB01111/A2A-RedPandaServer-Container/internal/auth"
 	"github.com/KB01111/A2A-RedPandaServer-Container/internal/config"
 	"github.com/KB01111/A2A-RedPandaServer-Container/internal/orchestrator"
 	appserver "github.com/KB01111/A2A-RedPandaServer-Container/internal/server"
+	"github.com/KB01111/A2A-RedPandaServer-Container/internal/storage/postgres"
 )
 
 func main() {
 	logger := slog.New(slog.NewJSONHandler(os.Stdout, nil))
+	if err := run(logger); err != nil {
+		logger.Error("server failed", "error", err)
+		os.Exit(1)
+	}
+}
+
+func run(logger *slog.Logger) error {
 	cfg, err := config.Load()
 	if err != nil {
-		logger.Error("invalid configuration", "error", err)
-		os.Exit(1)
+		return fmt.Errorf("invalid configuration: %w", err)
 	}
 	if cfg.Environment != "development" && cfg.Environment != "test" {
-		logger.Error("production dependencies are not configured", "environment", cfg.Environment)
-		os.Exit(1)
+		return fmt.Errorf("Redpanda dispatcher is not configured for %s", cfg.Environment)
 	}
 
-	handler, err := appserver.New(cfg, appserver.Dependencies{
+	shutdownSignals, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
+	defer stop()
+
+	dependencies := appserver.Dependencies{
 		Dispatcher: orchestrator.LoopbackDispatcher{},
 		Logger:     logger,
-	})
+	}
+	if cfg.OIDC.Enabled() {
+		verifier, err := auth.NewOIDCVerifier(shutdownSignals, cfg.OIDC)
+		if err != nil {
+			return fmt.Errorf("initialize OIDC verifier: %w", err)
+		}
+		dependencies.Authentication, err = auth.NewAuthenticator(verifier, cfg.OIDC.Issuer, cfg.OIDC.RequiredScopes)
+		if err != nil {
+			return fmt.Errorf("initialize authentication: %w", err)
+		}
+	}
+
+	if cfg.Database.URL != "" {
+		if dependencies.Authentication == nil {
+			return fmt.Errorf("DATABASE_URL requires OIDC authentication")
+		}
+		pool, err := postgres.OpenPool(shutdownSignals, postgres.PoolConfig{
+			DatabaseURL:       cfg.Database.URL,
+			PasswordFile:      cfg.Database.PasswordFile,
+			ApplicationName:   "bridge-a2a-server",
+			MaxConns:          cfg.Database.MaxConnections,
+			MinConns:          cfg.Database.MinConnections,
+			MaxConnLifetime:   cfg.Database.MaxConnectionLife,
+			MaxConnIdleTime:   cfg.Database.MaxConnectionIdle,
+			HealthCheckPeriod: cfg.Database.HealthCheckPeriod,
+		})
+		if err != nil {
+			return err
+		}
+		defer pool.Close()
+		if err := postgres.VerifySchema(shutdownSignals, pool); err != nil {
+			return fmt.Errorf("verify database schema: %w", err)
+		}
+		dependencies.TaskStore, err = postgres.NewStore(pool)
+		if err != nil {
+			return err
+		}
+	}
+
+	handler, err := appserver.New(cfg, dependencies)
 	if err != nil {
-		logger.Error("build server", "error", err)
-		os.Exit(1)
+		return fmt.Errorf("build server: %w", err)
 	}
 
 	server := &http.Server{
@@ -45,8 +94,6 @@ func main() {
 		MaxHeaderBytes:    1 << 20,
 	}
 
-	shutdownSignals, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
-	defer stop()
 	go func() {
 		<-shutdownSignals.Done()
 		ctx, cancel := context.WithTimeout(context.Background(), cfg.ShutdownTimeout)
@@ -58,7 +105,7 @@ func main() {
 
 	logger.Info("starting A2A server", "address", server.Addr, "environment", cfg.Environment)
 	if err := server.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
-		logger.Error("server stopped", "error", err)
-		os.Exit(1)
+		return fmt.Errorf("serve HTTP: %w", err)
 	}
+	return nil
 }

docs/architecture.md

@@ -26,6 +26,44 @@ The SDK does not itself enforce the `A2A-Version` service parameter. The HTTP
 stack therefore requires `A2A-Version: 1.0` for protocol calls and normalizes
 comma-separated `A2A-Extensions` values before handing requests to `a2asrv`.
 
+## Authentication and tenant boundary
+
+OIDC authentication wraps the A2A transport rather than running only as an SDK
+interceptor. This is deliberate: the SDK snapshots HTTP headers into request
+metadata before calling the executor. The outer middleware verifies a signed
+JWT, removes `Authorization`, `Proxy-Authorization`, and `Cookie`, and places a
+small verified identity in the request context. An SDK interceptor then sets
+the authenticated A2A user and injects the authoritative tenant claim into the
+typed request. A client-supplied tenant that differs from the signed claim is
+rejected.
+
+The accepted token profile is a signed JWT access token with the configured
+issuer, audience, asymmetric signing algorithm, `sub`, tenant, `exp`, and
+required scopes. Opaque tokens are not accepted. Discovery is performed once
+at startup and the long-lived verifier uses the provider's cached remote key
+set. An identity-provider outage therefore does not make liveness dependent on
+a network call for every request, although a previously unseen signing key
+still fails closed while the key endpoint is unavailable.
+
+## PostgreSQL task ownership
+
+The PostgreSQL adapter implements the SDK's `taskstore.Store` contract without
+changing A2A wire behavior. Every read and write is scoped by the verified OIDC
+issuer, tenant, and subject from `context.Context`; request JSON is never an
+authority source. Cross-scope lookups return not-found so they cannot be used
+as an existence oracle. Task IDs remain globally unique because the SDK's
+in-process event, work, and push stores are keyed only by task ID.
+
+The full canonical task is stored as JSONB with indexed projections for tenant,
+owner, context, state, status time, and update time. Updates use row locking and
+the SDK task version for optimistic concurrency. List operations use a
+repeatable-read snapshot and opaque keyset cursors.
+
+Schema changes are forward-only, checksummed, and protected by a PostgreSQL
+advisory lock. `cmd/migrate` owns DDL. The server only verifies that the schema
+is current and fails startup when it is behind; production deployments should
+use separate migration and runtime database roles.
+
 ## Runtime boundary
 
 The server is a stateless protocol and orchestration process. Agent workers,

go.mod

@@ -2,9 +2,19 @@ module github.com/KB01111/A2A-RedPandaServer-Container
 
 go 1.25.0
 
-require github.com/a2aproject/a2a-go/v2 v2.4.0
+require (
+	github.com/a2aproject/a2a-go/v2 v2.4.0
+	github.com/coreos/go-oidc/v3 v3.20.0
+	github.com/jackc/pgx/v5 v5.10.0
+)
 
 require (
+	github.com/go-jose/go-jose/v4 v4.1.4 // indirect
 	github.com/google/uuid v1.6.0 // indirect
+	github.com/jackc/pgpassfile v1.0.0 // indirect
+	github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect
+	github.com/jackc/puddle/v2 v2.2.2 // indirect
+	golang.org/x/oauth2 v0.36.0 // indirect
 	golang.org/x/sync v0.20.0 // indirect
+	golang.org/x/text v0.37.0 // indirect
 )

go.sum

@@ -1,10 +1,40 @@
 github.com/a2aproject/a2a-go/v2 v2.4.0 h1:da0iwA6voxLhXIw1B3GSAdrhs5UIa2v6m+v/Fix5qTw=
 github.com/a2aproject/a2a-go/v2 v2.4.0/go.mod h1:EghJ/rY9OCC6jme1z+otOBf+1YCVVCpMSPZU31t3hKY=
+github.com/coreos/go-oidc/v3 v3.20.0 h1:EtE0WIBHk03N+DqGkY4+UONzzZHk7amKt6IyNd7OsZE=
+github.com/coreos/go-oidc/v3 v3.20.0/go.mod h1:DYCf24+ncYi+XkIH97GY1+dqoRlbaSI26KVTCI9SrY4=
+github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
+github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/go-jose/go-jose/v4 v4.1.4 h1:moDMcTHmvE6Groj34emNPLs/qtYXRVcd6S7NHbHz3kA=
+github.com/go-jose/go-jose/v4 v4.1.4/go.mod h1:x4oUasVrzR7071A4TnHLGSPpNOm2a21K9Kf04k1rs08=
 github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8=
 github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU=
 github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
 github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
+github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM=
+github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg=
+github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 h1:iCEnooe7UlwOQYpKFhBabPMi4aNAfoODPEFNiAnClxo=
+github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761/go.mod h1:5TJZWKEWniPve33vlWYSoGYefn3gLQRzjfDlhSJ9ZKM=
+github.com/jackc/pgx/v5 v5.10.0 h1:VhSvgU2jSli8o3AqIEOTJr7rZwAEUVo4E4XhR94Zfr0=
+github.com/jackc/pgx/v5 v5.10.0/go.mod h1:mal1tBGAFfLHvZzaYh77YS/eC6IX9OWbRV1QIIM0Jn4=
+github.com/jackc/puddle/v2 v2.2.2 h1:PR8nw+E/1w0GLuRFSmiioY6UooMp6KJv0/61nB7icHo=
+github.com/jackc/puddle/v2 v2.2.2/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4=
+github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
+github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
+github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
+github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
+github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U=
+github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U=
 golang.org/x/mod v0.35.0 h1:Ww1D637e6Pg+Zb2KrWfHQUnH2dQRLBQyAtpr/haaJeM=
 golang.org/x/mod v0.35.0/go.mod h1:+GwiRhIInF8wPm+4AoT6L0FA1QWAad3OMdTRx4tFYlU=
+golang.org/x/oauth2 v0.36.0 h1:peZ/1z27fi9hUOFCAZaHyrpWG5lwe0RJEEEeH0ThlIs=
+golang.org/x/oauth2 v0.36.0/go.mod h1:YDBUJMTkDnJS+A4BP4eZBjCqtokkg1hODuPjwiGPO7Q=
 golang.org/x/sync v0.20.0 h1:e0PTpb7pjO8GAtTs2dQ6jYa5BWYlMuX047Dco/pItO4=
 golang.org/x/sync v0.20.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0=
+golang.org/x/text v0.37.0 h1:Cqjiwd9eSg8e0QAkyCaQTNHFIIzWtidPahFWR83rTrc=
+golang.org/x/text v0.37.0/go.mod h1:a5sjxXGs9hsn/AJVwuElvCAo9v8QYLzvavO5z2PiM38=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
+gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
+gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=