feat(policy): authorize authenticated principals

CHANGELOG.md

@@ -8,6 +8,8 @@ All notable changes to mcp-audit are documented in this file.
 
 - HTTP security principals with explicit local identity and optional static
   bearer authentication backed by constant-time token comparison.
+- Principal-aware policy selectors for subject, issuer, role, scope, method,
+  and operation name, plus minimal principal evidence in JSONL and SQLite.
 - HTTP proxy request-body and header limits, complete server timeouts, optional
   browser Origin validation, Host validation for DNS-rebinding protection, and
   `mcp_audit_http_request_rejections_total` metrics.

README.md

@@ -178,12 +178,12 @@ Prometheus metrics are available at `http://localhost:9091/metrics` by default.
 | `audit.rotation.interval` | empty | Optional time-based JSONL rotation interval: `hourly` or `daily`. Empty disables time-based rotation. |
 | `audit.rotation.max_age_days` | `0` | Delete JSONL archives whose filename rotation timestamp is older than this many days. `0` disables age retention. |
 | `middleware.rate_limit.enabled` | `true` | Enable per-client, per-tool token buckets. |
-| `middleware.rate_limit.requests_per_minute` | `60` | Allowed requests per minute per `(client_id, tool_name)`. |
+| `middleware.rate_limit.requests_per_minute` | `60` | Allowed requests per minute per authenticated `(client_id, tool_name)`. |
 | `middleware.redact.enabled` | `true` | Enable JSON key-based PII redaction. |
 | `middleware.redact.patterns` | sensitive keys | Case-insensitive key fragments to redact. |
 | `policy.enabled` | `false` | Enable synchronous allow/deny policy checks for `tools/call`. |
 | `policy.default_action` | `allow` | Fallback action when no policy rule matches: `allow` or `deny`. |
-| `policy.rules` | empty | Ordered first-match allow/deny rules for tool calls. |
+| `policy.rules` | empty | Ordered first-match allow/deny rules. Existing `client_id`, `server_id`, and `tool_name` selectors remain valid; optional principal selectors are `subject`, `issuer`, `role`, and `scope`, with generic operation selectors `method` and `name`. |
 | `dashboard.enabled` | `true` | Serve the dashboard. |
 | `dashboard.bind_address` | `127.0.0.1` | Dashboard listen address. Set explicitly, for example to `0.0.0.0`, only when the dashboard is protected by network controls or auth. |
 | `dashboard.port` | `9090` | Dashboard listen port. |

STABILITY.md

@@ -25,6 +25,8 @@ The following surfaces are covered by the stability policy starting at `v1.0.0`:
 - `proxy.forward_headers` is part of the stable configuration surface. Forwarded headers are passed verbatim to the trusted upstream HTTP MCP server, but HTTP headers are not recorded as dedicated fields in audit entries.
 - `proxy.bind_address` and the `proxy.http.*` request-limit, timeout, Origin, and Host validation keys are part of the stable configuration surface.
 - The `auth.mode` and `auth.static.*` keys are part of the stable configuration surface.
+- Principal-aware `policy.rules` selectors (`subject`, `issuer`, `role`,
+  `scope`, `method`, and `name`) are additive stable configuration keys.
 - The JSONL rotation keys (`audit.rotation.max_size_bytes`, `audit.rotation.max_files`, `audit.rotation.interval`, `audit.rotation.max_age_days`) are part of the stable configuration surface.
 
 ### CLI flags
@@ -35,7 +37,10 @@ The following surfaces are covered by the stability policy starting at `v1.0.0`:
 
 ### Audit entry JSON schema
 
-The fields recorded for each audit entry (`id`, `timestamp`, `direction`, `transport`, `method`, `request_id`, `tool_name`, `params`, `result`, `error`, `duration_ms`, `client_id`, `server_id`, `signature`) keep their names and types. New fields may be added in MINOR releases. Existing fields are not removed or renamed without a MAJOR bump.
+The fields recorded for each audit entry (`id`, `timestamp`, `direction`, `transport`, `method`, `request_id`, `tool_name`, `params`, `result`, `error`, `duration_ms`, `client_id`, `server_id`, `principal`, `signature`) keep their names and types. New fields may be added in MINOR releases. Existing fields are not removed or renamed without a MAJOR bump.
+
+The `principal` object contains only authenticated `subject`, `client_id`, and
+`issuer`. Raw JWT claims are never part of the audit schema.
 
 The signature is computed over `id + timestamp + method + tool_name + params`. Changing the signed field set requires a MAJOR bump because it invalidates existing signatures.
 

config.yaml

@@ -70,9 +70,13 @@ policy:
   default_action: allow
   rules:
     # - action: deny
-    #   client_id: claude-desktop
+    #   subject: alice@example.com
+    #   issuer: https://login.example.com
+    #   role: operator
+    #   scope: tools:delete
     #   server_id: filesystem
-    #   tool_name: delete_file
+    #   method: tools/call
+    #   name: delete_file
     #   reason: "Destructive filesystem operations are blocked"
 
 dashboard:

internal/audit/logger.go

@@ -24,6 +24,13 @@ type RPCError struct {
 	Data    json.RawMessage `json:"data,omitempty"`
 }
 
+// Principal is the minimal authenticated identity retained as audit evidence.
+type Principal struct {
+	Subject  string `json:"subject"`
+	ClientID string `json:"client_id"`
+	Issuer   string `json:"issuer"`
+}
+
 // Entry is a single audited JSON-RPC exchange or message.
 type Entry struct {
 	ID         string          `json:"id"`
@@ -39,6 +46,7 @@ type Entry struct {
 	DurationMs int64           `json:"duration_ms"`
 	ClientID   string          `json:"client_id"`
 	ServerID   string          `json:"server_id"`
+	Principal  *Principal      `json:"principal,omitempty"`
 	Signature  string          `json:"signature"`
 }
 

internal/audit/storage/sqlite.go

@@ -56,6 +56,7 @@ func (s *SQLiteStore) init(ctx context.Context) error {
 			duration_ms INTEGER NOT NULL,
 			client_id TEXT NOT NULL,
 			server_id TEXT NOT NULL,
+			principal TEXT,
 			signature TEXT
 		)`,
 		`CREATE INDEX IF NOT EXISTS idx_audit_entries_timestamp ON audit_entries(timestamp)`,
@@ -72,6 +73,10 @@ func (s *SQLiteStore) init(ctx context.Context) error {
 		!strings.Contains(err.Error(), "duplicate column name") {
 		return fmt.Errorf("audit: sqlite: migrate request_id: %w", err)
 	}
+	if _, err := s.db.ExecContext(ctx, `ALTER TABLE audit_entries ADD COLUMN principal TEXT`); err != nil &&
+		!strings.Contains(err.Error(), "duplicate column name") {
+		return fmt.Errorf("audit: sqlite: migrate principal: %w", err)
+	}
 	return nil
 }
 
@@ -93,8 +98,8 @@ func (s *SQLiteStore) AppendBatch(entries []audit.Entry) error {
 	}
 	stmt, err := tx.Prepare(`INSERT INTO audit_entries (
 		id, timestamp, direction, transport, method, request_id, tool_name, params, result, error,
-		duration_ms, client_id, server_id, signature
-	) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`)
+		duration_ms, client_id, server_id, principal, signature
+	) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`)
 	if err != nil {
 		_ = tx.Rollback()
 		return fmt.Errorf("audit: sqlite: prepare batch: %w", err)
@@ -110,6 +115,14 @@ func (s *SQLiteStore) AppendBatch(entries []audit.Entry) error {
 		if entry.Error == nil {
 			errorJSON = nil
 		}
+		principalJSON, err := json.Marshal(entry.Principal)
+		if err != nil {
+			_ = tx.Rollback()
+			return fmt.Errorf("audit: sqlite: marshal principal: %w", err)
+		}
+		if entry.Principal == nil {
+			principalJSON = nil
+		}
 		_, err = stmt.Exec(
 			entry.ID,
 			entry.Timestamp.UTC().Format(time.RFC3339Nano),
@@ -124,6 +137,7 @@ func (s *SQLiteStore) AppendBatch(entries []audit.Entry) error {
 			entry.DurationMs,
 			entry.ClientID,
 			entry.ServerID,
+			string(principalJSON),
 			entry.Signature,
 		)
 		if err != nil {
@@ -140,7 +154,7 @@ func (s *SQLiteStore) AppendBatch(entries []audit.Entry) error {
 // Query returns recent entries matching filter.
 func (s *SQLiteStore) Query(filter audit.QueryFilter) ([]audit.Entry, error) {
 	rows, err := s.db.Query(`SELECT id, timestamp, direction, transport, method, request_id, tool_name, params, result, error,
-		duration_ms, client_id, server_id, signature
+		duration_ms, client_id, server_id, principal, signature
 		FROM audit_entries
 		ORDER BY timestamp DESC
 		LIMIT 10000`)
@@ -152,7 +166,7 @@ func (s *SQLiteStore) Query(filter audit.QueryFilter) ([]audit.Entry, error) {
 	var entries []audit.Entry
 	for rows.Next() {
 		var entry audit.Entry
-		var timestamp, params, result, rpcErr sql.NullString
+		var timestamp, params, result, rpcErr, principal sql.NullString
 		if err := rows.Scan(
 			&entry.ID,
 			&timestamp,
@@ -167,6 +181,7 @@ func (s *SQLiteStore) Query(filter audit.QueryFilter) ([]audit.Entry, error) {
 			&entry.DurationMs,
 			&entry.ClientID,
 			&entry.ServerID,
+			&principal,
 			&entry.Signature,
 		); err != nil {
 			return nil, fmt.Errorf("audit: sqlite: scan: %w", err)
@@ -189,6 +204,12 @@ func (s *SQLiteStore) Query(filter audit.QueryFilter) ([]audit.Entry, error) {
 				entry.Error = &decoded
 			}
 		}
+		if principal.Valid && principal.String != "" && principal.String != "null" {
+			var decoded audit.Principal
+			if err := json.Unmarshal([]byte(principal.String), &decoded); err == nil {
+				entry.Principal = &decoded
+			}
+		}
 		if audit.MatchFilter(entry, filter) {
 			entries = append(entries, entry)
 		}

internal/audit/storage/sqlite_test.go

@@ -372,6 +372,21 @@ func TestSQLiteStorePersistsRequestID(t *testing.T) {
 	}
 }
 
+func TestSQLiteStorePersistsPrincipalProjection(t *testing.T) {
+	t.Parallel()
+	store := newSQLiteStore(t)
+	want := &audit.Principal{Subject: "alice", ClientID: "client-1", Issuer: "https://issuer.example.com"}
+	mustSQLiteAppend(t, store, audit.Entry{ID: "principal-1", ClientID: "client-1", ServerID: "s1", Principal: want})
+
+	entries, err := store.Query(audit.QueryFilter{})
+	if err != nil {
+		t.Fatalf("Query: %v", err)
+	}
+	if len(entries) != 1 || entries[0].Principal == nil || *entries[0].Principal != *want {
+		t.Fatalf("principal = %#v", entries)
+	}
+}
+
 func TestNewSQLiteStoreCreatesParentDirectory(t *testing.T) {
 	t.Parallel()
 	path := filepath.Join(t.TempDir(), "sub", "dir", "audit.db")

internal/mcp/headers.go

@@ -0,0 +1,50 @@
+package mcp
+
+import (
+	"fmt"
+	"net/http"
+	"strings"
+)
+
+const (
+	HeaderMethod          = "Mcp-Method"
+	HeaderName            = "Mcp-Name"
+	HeaderProtocolVersion = "Mcp-Protocol-Version"
+)
+
+type requestHeaders struct {
+	method   string
+	name     string
+	revision string
+}
+
+func inspectHeaders(headers http.Header) (requestHeaders, error) {
+	method, err := singleHeader(headers, HeaderMethod)
+	if err != nil {
+		return requestHeaders{}, err
+	}
+	name, err := singleHeader(headers, HeaderName)
+	if err != nil {
+		return requestHeaders{}, err
+	}
+	revision, err := singleHeader(headers, HeaderProtocolVersion)
+	if err != nil {
+		return requestHeaders{}, err
+	}
+	return requestHeaders{method: method, name: name, revision: revision}, nil
+}
+
+func singleHeader(headers http.Header, name string) (string, error) {
+	values := headers.Values(name)
+	if len(values) == 0 {
+		return "", nil
+	}
+	if len(values) != 1 {
+		return "", fmt.Errorf("mcp: header %s must occur at most once", name)
+	}
+	value := strings.TrimSpace(values[0])
+	if value == "" || value != values[0] {
+		return "", fmt.Errorf("mcp: header %s must be non-empty without surrounding whitespace", name)
+	}
+	return value, nil
+}

internal/mcp/message.go

@@ -0,0 +1,85 @@
+package mcp
+
+import (
+	"bytes"
+	"encoding/json"
+	"fmt"
+	"io"
+)
+
+const maxBatchMessages = 1024
+
+// Message contains the JSON-RPC request fields needed for gateway inspection.
+type Message struct {
+	JSONRPC string          `json:"jsonrpc,omitempty"`
+	ID      json.RawMessage `json:"id,omitempty"`
+	Method  string          `json:"method,omitempty"`
+	Params  json.RawMessage `json:"params,omitempty"`
+}
+
+// DecodeMessages parses a single JSON-RPC request or a bounded request batch.
+func DecodeMessages(raw []byte) ([]Message, error) {
+	decoder := json.NewDecoder(bytes.NewReader(raw))
+	decoder.UseNumber()
+	var payload json.RawMessage
+	if err := decoder.Decode(&payload); err != nil {
+		return nil, fmt.Errorf("mcp: decode request: %w", err)
+	}
+	if err := ensureJSONEOF(decoder); err != nil {
+		return nil, err
+	}
+	payload = bytes.TrimSpace(payload)
+	if len(payload) == 0 {
+		return nil, fmt.Errorf("mcp: request is empty")
+	}
+	if payload[0] != '[' {
+		message, err := decodeMessage(payload)
+		if err != nil {
+			return nil, err
+		}
+		return []Message{message}, nil
+	}
+	var rawMessages []json.RawMessage
+	if err := json.Unmarshal(payload, &rawMessages); err != nil {
+		return nil, fmt.Errorf("mcp: decode request batch: %w", err)
+	}
+	if len(rawMessages) == 0 {
+		return nil, fmt.Errorf("mcp: request batch is empty")
+	}
+	if len(rawMessages) > maxBatchMessages {
+		return nil, fmt.Errorf("mcp: request batch exceeds %d messages", maxBatchMessages)
+	}
+	messages := make([]Message, 0, len(rawMessages))
+	for index, rawMessage := range rawMessages {
+		message, err := decodeMessage(rawMessage)
+		if err != nil {
+			return nil, fmt.Errorf("mcp: request batch item %d: %w", index, err)
+		}
+		messages = append(messages, message)
+	}
+	return messages, nil
+}
+
+func decodeMessage(raw json.RawMessage) (Message, error) {
+	if trimmed := bytes.TrimSpace(raw); len(trimmed) == 0 || trimmed[0] != '{' {
+		return Message{}, fmt.Errorf("request message must be an object")
+	}
+	var message Message
+	if err := json.Unmarshal(raw, &message); err != nil {
+		return Message{}, fmt.Errorf("decode request message: %w", err)
+	}
+	if message.Method == "" {
+		return Message{}, fmt.Errorf("request method is required")
+	}
+	return message, nil
+}
+
+func ensureJSONEOF(decoder *json.Decoder) error {
+	var trailing json.RawMessage
+	if err := decoder.Decode(&trailing); err == io.EOF {
+		return nil
+	} else if err != nil {
+		return fmt.Errorf("mcp: decode trailing data: %w", err)
+	}
+	return fmt.Errorf("mcp: multiple JSON values are not allowed")
+}

internal/mcp/message_test.go

@@ -0,0 +1,52 @@
+package mcp
+
+import (
+	"fmt"
+	"strings"
+	"testing"
+)
+
+func TestDecodeMessagesAcceptsSingleAndBatch(t *testing.T) {
+	single, err := DecodeMessages([]byte(`{"jsonrpc":"2.0","id":1,"method":"tools/list"}`))
+	if err != nil || len(single) != 1 || single[0].Method != "tools/list" {
+		t.Fatalf("single = %#v, err = %v", single, err)
+	}
+	batch, err := DecodeMessages([]byte(`[{"jsonrpc":"2.0","id":1,"method":"tools/list"},{"jsonrpc":"2.0","method":"logging/setLevel"}]`))
+	if err != nil || len(batch) != 2 {
+		t.Fatalf("batch = %#v, err = %v", batch, err)
+	}
+}
+
+func TestDecodeMessagesRejectsInvalidPayloads(t *testing.T) {
+	cases := []string{
+		``,
+		`null`,
+		`[]`,
+		`{"jsonrpc":"2.0","id":1}`,
+		`{"jsonrpc":"2.0","method":"ping"} {"method":"ping"}`,
+		`[1]`,
+	}
+	for _, body := range cases {
+		if _, err := DecodeMessages([]byte(body)); err == nil {
+			t.Fatalf("expected error for %q", body)
+		}
+	}
+}
+
+func TestDecodeMessagesBoundsBatchCardinality(t *testing.T) {
+	items := make([]string, maxBatchMessages+1)
+	for index := range items {
+		items[index] = fmt.Sprintf(`{"jsonrpc":"2.0","id":%d,"method":"ping"}`, index)
+	}
+	body := "[" + strings.Join(items, ",") + "]"
+	if _, err := DecodeMessages([]byte(body)); err == nil {
+		t.Fatal("expected oversized batch error")
+	}
+}
+
+func TestInspectRequestRejectsBatchHeaders(t *testing.T) {
+	body := []byte(`[{"jsonrpc":"2.0","id":1,"method":"tools/list"},{"jsonrpc":"2.0","id":2,"method":"prompts/list"}]`)
+	if _, err := InspectRequest(nil, body); err == nil {
+		t.Fatal("expected batch inspection error")
+	}
+}