Phase 1, kafka, oauth, saml mdq validation.

.vscode/settings.json

@@ -97,6 +97,7 @@
         "netcat",
         "newrequest",
         "NEWTAG",
+        "nolint",
         "nosec",
         "NOSONAR",
         "Numerify",
@@ -142,16 +143,19 @@
         "samlsp",
         "samltypes",
         "sarama",
+        "SASL",
         "Satosa",
         "SATOSAV",
         "schac",
+        "SCRAMSHA",
         "SDJWT",
         "sdjwtvc",
         "sdktrace",
         "secp",
         "securego",
         "semconv",
         "setnx",
+        "setnxttl",
         "sexp",
         "shortuuid",
         "simplequeue",
@@ -192,6 +196,7 @@
         "vulncheck",
         "wrongverifier",
         "wwwallet",
+        "XDGSCRAM",
         "zapr",
         "zenghongtu",
         "zenor",

internal/apigw/apiv1/handlers_oauth.go

@@ -182,28 +182,10 @@ func (c *Client) OAuthToken(ctx context.Context, req *openid4vci.TokenRequest) (
 		code = req.PreAuthorizedCode
 	}
 
-	// Look up the client to enforce type-specific requirements (client_id is optional for pre-auth flow)
-	// When client_assertion is provided (private_key_jwt auth), extract client_id from the assertion's sub claim.
-	// SECURITY: ExtractClientIDFromAssertion only decodes the JWT payload — it does NOT verify the signature.
-	// The extracted client_id is used for both client config lookup (Clients.Get) AND client binding
-	// verification (WalletClientID check). Without signature verification, an attacker could forge
-	// the sub claim to impersonate another client.
-	// Full JWT assertion verification (signature, aud, exp, jti) is NOT yet implemented.
-	// TODO(security): Implement RFC 7523 private_key_jwt assertion verification
-	// (signature, aud, exp, jti) before enabling client_assertion in production.
-	// Tracked as a known risk — see risk register SID-RISK-CLIENT-ASSERTION.
+	// When client_assertion is provided (private_key_jwt auth), verify the assertion
+	// signature per RFC 7523 and extract client_id from the sub claim.
 	clientID := req.ClientID
 	if req.ClientAssertion != "" {
-		// SECURITY: client_assertion signature is NOT verified — only the payload is decoded.
-		// Reject unless explicitly opted in via allow_unverified_client_assertion config flag.
-		// Full RFC 7523 verification (signature, aud, exp, jti) must be implemented before
-		// removing this flag. Tracked in risk register SID-RISK-CLIENT-ASSERTION.
-		if !c.cfg.APIGW.Delivery.OpenID4VCI.AllowUnverifiedClientAssertion {
-			return nil, oauth2.NewOAuthError(oauth2.ErrCodeInvalidRequest,
-				"client_assertion is not supported (RFC 7523 verification not implemented)", 400)
-		}
-		c.log.Warn("accepting unverified client_assertion (allow_unverified_client_assertion is enabled) — signature verification not implemented",
-			"client_assertion_type", req.ClientAssertionType)
 		if req.ClientAssertionType == "" {
 			return nil, oauth2.NewOAuthError(oauth2.ErrCodeInvalidRequest,
 				"client_assertion_type is required when client_assertion is provided", 400)
@@ -212,18 +194,73 @@ func (c *Client) OAuthToken(ctx context.Context, req *openid4vci.TokenRequest) (
 			return nil, oauth2.NewOAuthError(oauth2.ErrCodeInvalidRequest,
 				fmt.Sprintf("unsupported client_assertion_type %q; expected urn:ietf:params:oauth:client-assertion-type:jwt-bearer", req.ClientAssertionType), 400)
 		}
-		sub, err := oauth2.ExtractClientIDFromAssertion(req.ClientAssertion)
-		if err != nil {
-			c.log.Error(err, "failed to extract client_id from client_assertion")
-			return nil, oauth2.NewOAuthErrorWithCause(oauth2.ErrCodeInvalidClient,
-				"Invalid client assertion", 401, err)
-		}
-		if clientID != "" && clientID != sub {
-			return nil, oauth2.NewOAuthError(oauth2.ErrCodeInvalidClient,
-				"client_id does not match assertion subject", 401)
+
+		if c.cfg.APIGW.Delivery.OpenID4VCI.AllowUnverifiedClientAssertion {
+			// CONFORMANCE TESTING ONLY: accept assertion without signature verification.
+			c.log.Warn("accepting unverified client_assertion (allow_unverified_client_assertion is enabled)",
+				"client_assertion_type", req.ClientAssertionType)
+			sub, err := oauth2.ExtractClientIDFromAssertion(req.ClientAssertion)
+			if err != nil {
+				c.log.Error(err, "failed to extract client_id from client_assertion")
+				return nil, oauth2.NewOAuthErrorWithCause(oauth2.ErrCodeInvalidClient,
+					"Invalid client assertion", 401, err)
+			}
+			if clientID != "" && clientID != sub {
+				return nil, oauth2.NewOAuthError(oauth2.ErrCodeInvalidClient,
+					"client_id does not match assertion subject", 401)
+			}
+			clientID = sub
+		} else {
+			// RFC 7523 full verification path: extract sub for client lookup, then verify signature.
+			sub, err := oauth2.ExtractClientIDFromAssertion(req.ClientAssertion)
+			if err != nil {
+				c.log.Error(err, "failed to extract client_id from client_assertion")
+				return nil, oauth2.NewOAuthErrorWithCause(oauth2.ErrCodeInvalidClient,
+					"Invalid client assertion", 401, err)
+			}
+			if clientID != "" && clientID != sub {
+				return nil, oauth2.NewOAuthError(oauth2.ErrCodeInvalidClient,
+					"client_id does not match assertion subject", 401)
+			}
+			clientID = sub
+
+			// Look up client to get JWKS URI for signature verification
+			oauthClientForVerify, err := c.cfg.APIGW.Delivery.OpenID4VCI.Clients.Get(clientID)
+			if err != nil {
+				return nil, oauth2.NewOAuthErrorWithCause(oauth2.ErrCodeInvalidClient,
+					"Client authentication failed", 401, err)
+			}
+
+			verifier := &oauth2.ClientAssertionVerifier{
+				TokenEndpoint: c.cfg.APIGW.Delivery.OpenID4VCI.TokenEndpoint,
+				JWKSCache:     c.cacheService.JWKS,
+				JTICheck: func(jti string, exp time.Time) error {
+					// Scope by clientID to prevent cross-client collisions;
+					// use time.Until(exp) as TTL so entries expire with the assertion.
+					cacheKey := "client_assertion:" + clientID + ":" + jti
+					// Add the same leeway as the JWT verifier to tolerate small clock skews.
+					ttl := time.Until(exp.Add(30 * time.Second))
+					if ttl <= 0 {
+						return errors.New("client_assertion jti has already expired")
+					}
+					unique, err := c.cacheService.DPopJTI.SetNXWithTTL(ctx, cacheKey, true, ttl)
+					if err != nil {
+						return fmt.Errorf("jti cache error: %w", err)
+					}
+					if !unique {
+						return errors.New("client_assertion jti already used")
+					}
+					return nil
+				},
+			}
+			assertionClaims, err := verifier.Verify(ctx, req.ClientAssertion, oauthClientForVerify)
+			if err != nil {
+				c.log.Error(err, "client_assertion verification failed", "client_id", clientID)
+				return nil, oauth2.NewOAuthErrorWithCause(oauth2.ErrCodeInvalidClient,
+					"Client assertion verification failed", 401, err)
+			}
+			c.log.Debug("client_assertion verified", "client_id", clientID, "jti", assertionClaims.JTI)
 		}
-		clientID = sub
-		c.log.Debug("OAuthToken: resolved client_id from client_assertion", "client_id", clientID)
 	} else if req.ClientAssertionType != "" {
 		return nil, oauth2.NewOAuthError(oauth2.ErrCodeInvalidRequest,
 			"client_assertion_type provided without client_assertion", 400)

internal/apigw/auth_providers/samlsp/service.go

@@ -29,7 +29,7 @@
 }
 
 // New creates a new SAML service
 func New(ctx context.Context, cfg *model.SAMLSP, sessionCache pkgcache.Cache[*Session], log *logger.Log) (*Service, error) {
 	if !cfg.Enable {
 		log.Info("SAML support disabled")
 		return nil, nil
@@ -94,6 +94,25 @@
 			"cert_path", cfg.MetadataSigningCertPath)
 	}
 
+	// Determine if the metadata source is MDQ or URL-based (requires signature verification)
+	requiresRemoteFetch := cfg.MDQServer != "" ||
+		(cfg.StaticIDPMetadata != nil && cfg.StaticIDPMetadata.MetadataURL != "")
+
+	// Enforce metadata signature verification for MDQ/URL sources unless explicitly opted out
+	if requiresRemoteFetch && metadataSigningCert == nil && !cfg.AllowUnsignedMetadata {
+		return nil, fmt.Errorf("metadata_signing_cert_path is required when using MDQ or URL-based metadata sources " +
+			"(set allow_unsigned_metadata: true to override — NOT RECOMMENDED for production)")
+	}
+	if requiresRemoteFetch && metadataSigningCert == nil && cfg.AllowUnsignedMetadata {
+		s.log.Warn("INSECURE: accepting unsigned metadata from remote sources (allow_unsigned_metadata is enabled)")
+	}
+
+	// Warn for local file metadata without signature verification (allowed but noted)
+	if cfg.StaticIDPMetadata != nil && cfg.StaticIDPMetadata.MetadataPath != "" && metadataSigningCert == nil {
+		s.log.Warn("static IdP metadata loaded from local file without signature verification",
+			"path", cfg.StaticIDPMetadata.MetadataPath)
+	}
+
 	// Initialize MDQ client (either MDQ or static mode)
 	if cfg.StaticIDPMetadata != nil {
 		// Static IdP mode

internal/apigw/outbound/kafka_message_publisher.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"encoding/json"
 	"errors"
+	"fmt"
 	"reflect"
 
 	"github.com/SUNET/vc/internal/apigw/apiv1"
@@ -22,7 +23,10 @@ type kafkaMessageProducer struct {
 
 // New creates a new instance of a kafka event publisher used by apigw
 func New(ctx context.Context, cfg *model.Cfg, tracer *trace.Tracer, log *logger.Log) (apiv1.EventPublisher, error) {
-	saramaConfig := kafka.CommonProducerConfig(cfg)
+	saramaConfig, err := kafka.CommonProducerConfig(cfg)
+	if err != nil {
+		return nil, fmt.Errorf("kafka producer security config: %w", err)
+	}
 	client, err := kafka.NewSyncProducerClient(ctx, saramaConfig, cfg, tracer, log.New("kafka_message_producer_client"))
 	if err != nil {
 		return nil, err

internal/verifier/apiv1/handlers_verification.go

@@ -26,7 +26,7 @@ type VerificationRequestObjectRequest struct {
 }
 
 func (c *Client) VerificationRequestObject(ctx context.Context, req *VerificationRequestObjectRequest) (string, error) {
-	c.log.Debug("Verification request object", "req", req)
+	c.log.Debug("Verification request object", "id", req.ID)
 
 	// Query by RequestObjectID since that's what the wallet sends via ?id= parameter
 	authorizationContext, err := c.cacheService.AuthContext.Get(ctx, &cache.AuthorizationContext{
@@ -50,7 +50,7 @@ func (c *Client) VerificationRequestObject(ctx context.Context, req *Verificatio
 		return "", err
 	}
 
-	c.log.Debug("Signed JWT", "jwt", signedJWT)
+	c.log.Debug("Signed JWT created", "requestObjectID", authorizationContext.RequestObjectID)
 
 	return signedJWT, nil
 }
@@ -103,7 +103,7 @@ func (c *Client) VerificationDirectPost(ctx context.Context, req *VerificationDi
 		return nil, err
 	}
 
-	c.log.Debug("directPost", "vpResponse", vpResponse)
+	c.log.Debug("directPost", "state", vpResponse.State, "credential_count", len(vpResponse.VPToken))
 
 	// Get authorization context by state
 	authCtx, err := c.cacheService.AuthContext.Get(ctx, &cache.AuthorizationContext{State: vpResponse.State})

pkg/cache/generic.go

@@ -36,6 +36,12 @@ type Cache[V any] interface {
 	// "already exists" from operational errors.
 	SetNX(ctx context.Context, key string, value V) (bool, error)
 
+	// SetNXWithTTL stores a value only if the key does not already exist (atomic),
+	// using a custom TTL instead of the default. Returns true if the value was set,
+	// false if the key already existed.
+	// If ttl <= 0, implementations MUST fall back to SetNX (default TTL).
+	SetNXWithTTL(ctx context.Context, key string, value V, ttl time.Duration) (bool, error)
+
 	// SetWithTTL stores a value with a custom TTL, overriding the default.
 	SetWithTTL(ctx context.Context, key string, value V, ttl time.Duration)
 

pkg/cache/generic_memory.go

@@ -46,6 +46,17 @@ func (m *MemoryCache[V]) SetNX(_ context.Context, key string, value V) (bool, er
 	return !found, nil
 }
 
+// SetNXWithTTL stores a value only if the key does not already exist, using a custom TTL.
+// Returns true if the value was set, false if the key already existed.
+// If ttl <= 0, falls back to SetNX (default TTL).
+func (m *MemoryCache[V]) SetNXWithTTL(ctx context.Context, key string, value V, ttl time.Duration) (bool, error) {
+	if ttl <= 0 {
+		return m.SetNX(ctx, key, value)
+	}
+	_, found := m.cache.GetOrSet(key, value, ttlcache.WithTTL[string, V](ttl))
+	return !found, nil
+}
+
 // SetWithTTL stores a value with a custom TTL.
 func (m *MemoryCache[V]) SetWithTTL(_ context.Context, key string, value V, ttl time.Duration) {
 	m.cache.Set(key, value, ttl)

pkg/cache/generic_mongo.go

@@ -148,6 +148,29 @@ func (m *MongoCache[V]) SetNX(ctx context.Context, key string, value V) (bool, e
 	return true, nil
 }
 
+// SetNXWithTTL stores a value only if the key does not already exist (atomic),
+// using a custom TTL approximated via created_at shifting (same as SetWithTTL).
+// If ttl <= 0, falls back to SetNX (default TTL).
+func (m *MongoCache[V]) SetNXWithTTL(ctx context.Context, key string, value V, ttl time.Duration) (bool, error) {
+	if ttl <= 0 {
+		return m.SetNX(ctx, key, value)
+	}
+	shift := m.ttl - ttl
+	createdAt := time.Now().Add(-shift)
+	entry, err := m.marshalEntry(key, value, createdAt)
+	if err != nil {
+		return false, fmt.Errorf("mongo cache setnxttl marshal failed (cache=%s): %w", m.collection, err)
+	}
+	_, err = m.coll.InsertOne(ctx, entry)
+	if err != nil {
+		if mongo.IsDuplicateKeyError(err) {
+			return false, nil
+		}
+		return false, fmt.Errorf("mongo cache setnxttl failed (cache=%s): %w", m.collection, err)
+	}
+	return true, nil
+}
+
 // SetWithTTL stores a value with a custom TTL.
 // MongoDB TTL indexes are collection-wide, so per-entry TTL is approximated
 // by shifting created_at: the document expires when