valkey

package
v0.3.2 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jun 12, 2026 License: Apache-2.0 Imports: 22 Imported by: 0

Documentation

Overview

Package valkey provides a Valkey storage backend for the mcp-oauth library.

Valkey is a high-performance key-value store that is wire-compatible with Redis. This package implements all storage interfaces required by the mcp-oauth library, making it suitable for production deployments that require:

  • Distributed storage for horizontal scaling
  • Persistence across server restarts
  • Automatic TTL-based expiration
  • High availability with clustering

Implemented Interfaces

The Store type implements all required storage interfaces:

Key Schema

All keys use a configurable prefix (default "mcp:") to avoid conflicts with other applications sharing the same Valkey instance: 

{prefix}:token:{userID}           -> JSON(oauth2.Token)
{prefix}:userinfo:{userID}        -> JSON(UserInfo)
{prefix}:refresh:{token}          -> userID (with TTL)
{prefix}:refresh:meta:{token}     -> JSON(familyMetadata)
{prefix}:client:{clientID}        -> JSON(Client)
{prefix}:client:ip:{ip}           -> count (with TTL)
{prefix}:state:{stateID}          -> JSON(AuthorizationState)
{prefix}:state:provider:{state}   -> stateID (for reverse lookup)
{prefix}:code:{code}              -> JSON(AuthorizationCode)
{prefix}:meta:{tokenID}           -> JSON(TokenMetadata)
{prefix}:userclient:{uid}:{cid}   -> SET of tokenIDs
{prefix}:family:{familyID}        -> SET of refresh tokens in family

Atomic Operations

OAuth 2.1 requires certain operations to be atomic to prevent security issues:

  • AtomicCheckAndMarkAuthCodeUsed: Prevents authorization code replay attacks
  • AtomicGetAndDeleteRefreshToken: Prevents refresh token reuse attacks

These operations use Lua scripts to ensure atomicity in Valkey, providing the same security guarantees as the in-memory implementation but with distributed storage benefits.

Configuration

Basic usage: 

store, err := valkey.New(valkey.Config{
    Address:   "localhost:6379",
    KeyPrefix: "mcp:",
})

With TLS: 

store, err := valkey.New(valkey.Config{
    Address:   "valkey.example.com:6379",
    Password:  os.Getenv("VALKEY_PASSWORD"),
    TLS:       &tls.Config{MinVersion: tls.VersionTLS12},
    KeyPrefix: "mcp:",
})

Security Considerations

  • All tokens are stored with TTLs to prevent unbounded growth
  • Lua scripts ensure atomic operations for security-critical flows
  • Constant-time bcrypt comparison prevents timing attacks in client validation
  • TLS support enables encrypted connections to Valkey servers
  • Family metadata is retained for configurable period for security forensics
  • Optional token encryption at rest via WithEncryptor() using AES-256-GCM
  • Input size validation prevents DoS attacks via oversized payloads
  • Generic error messages prevent information leakage

Token Encryption at Rest

Sensitive oauth2.Token fields (AccessToken, RefreshToken) can be encrypted before storing in Valkey: 

key, _ := security.GenerateKey()
encryptor, _ := security.NewEncryptor(key)
store, err := valkey.New(cfg, valkey.WithEncryptor(encryptor))

When enabled, tokens are encrypted with AES-256-GCM before storage and automatically decrypted when retrieved.

Best Practices

  • Always use TLS in production environments
  • Set strong passwords for Valkey authentication
  • Enable token encryption at rest for sensitive deployments
  • Use dedicated Valkey instances or databases for OAuth storage
  • Monitor key count and memory usage for potential DoS attacks
  • Configure appropriate TTLs based on your security requirements

Index

Constants

View Source 
const (
	// DefaultKeyPrefix is the default prefix for all Valkey keys
	DefaultKeyPrefix = "mcp:"

	// DefaultRevokedFamilyRetentionDays is the default retention period for revoked token families
	DefaultRevokedFamilyRetentionDays = 90

	// DefaultRefreshTokenTTL is the default TTL for provider tokens that have
	// a refresh token. This prevents orphaned keys from living indefinitely
	// when the refresh token is consumed or expires without explicit cleanup.
	DefaultRefreshTokenTTL = 90 * 24 * time.Hour

	// MaxInputLength is the maximum allowed length for any caller-supplied string
	// (tokenID, userID, clientID, refreshToken, familyID). Key components are
	// separately bounded by hashing (see hashKeyComponent).
	MaxInputLength = maxInputValueLength

	// MaxTokenLength and MaxIDLength are retained for API compatibility.
	// Deprecated: use MaxInputLength. These values no longer reflect the enforced
	// limit (MaxInputLength = 16 KiB); callers or downstream consumers using them
	// for pre-validation must remove that check — it will falsely reject valid JWTs
	// and long IdP subjects (e.g. Dex Kubernetes-connector base64-protobuf subjects).
	MaxTokenLength = 512
	MaxIDLength    = 256

	// DefaultMaxTokenDataSize is the default ceiling on the serialized token
	// written to Valkey, applied after AES-256-GCM + base64 expansion of the
	// encrypted-at-rest fields (AccessToken, RefreshToken, id_token). 600 KiB
	// admits raw id_tokens up to ~440 KiB once encrypted, covering enterprise
	// OIDC populations whose JWTs embed a full `groups` claim. Override per
	// store via [Config.MaxTokenDataSize].
	DefaultMaxTokenDataSize = 600 * 1024

	// MinMaxTokenDataSize is the lower bound accepted for
	// [Config.MaxTokenDataSize]. A smaller ceiling cannot fit an encrypted
	// minimal OIDC id_token with even a short groups claim.
	MinMaxTokenDataSize = 64 * 1024

	// MaxMaxTokenDataSize is the upper bound accepted for
	// [Config.MaxTokenDataSize]. Values larger than this are rejected to
	// preserve a DoS budget at the SaveToken boundary.
	MaxMaxTokenDataSize = 8 * 1024 * 1024
)

Variables

View Source 
var (

	// ErrInputTooLarge is returned when a caller-supplied value exceeds maxInputValueLength.
	// Callers can use errors.Is to distinguish this from storage-layer errors.
	ErrInputTooLarge = fmt.Errorf("input exceeds maximum allowed size")
)

Validation error messages (generic to prevent information leakage)

Functions

func NewDPoPReplayCache added in v0.2.164 

func NewDPoPReplayCache(client vk.Client, keyPrefix string) server.DPoPReplayCache

NewDPoPReplayCache returns a server.DPoPReplayCache backed by Valkey. keyPrefix is prepended to every JTI key; use it to namespace across environments or tenants (e.g. "muster:dpop:").

Types

type Config 

type Config struct {
	// Address is the Valkey server address (required), e.g., "localhost:6379"
	Address string

	// Password is the optional password for Valkey authentication
	Password string

	// DB is the optional database number (default 0)
	DB int

	// KeyPrefix is the prefix for all keys (default "mcp:")
	KeyPrefix string

	// TLS is the optional TLS configuration for encrypted connections
	TLS *tls.Config

	// Logger is the optional structured logger (default: slog.Default())
	Logger *slog.Logger

	// RevokedFamilyRetentionDays is the retention period for revoked token family metadata
	// Used for security forensics and auditing. Default: 90 days
	RevokedFamilyRetentionDays int

	// RefreshTokenTTL is the TTL applied to provider tokens that have a refresh
	// token. This should match the MCP server's refresh token lifetime so that
	// Valkey automatically evicts orphaned provider tokens. Default: 90 days
	RefreshTokenTTL time.Duration

	// MaxTokenDataSize bounds the serialized token written by [Store.SaveToken],
	// applied after encryption and JSON framing. Zero selects
	// [DefaultMaxTokenDataSize]. Values outside [[MinMaxTokenDataSize],
	// [MaxMaxTokenDataSize]] cause [New] to return an error.
	MaxTokenDataSize int
}

Config holds configuration for the Valkey storage backend.

type Option added in v0.2.160 

type Option func(*Store)

Option configures a Store at construction time.

func WithEncryptor added in v0.2.160 

func WithEncryptor(enc *security.Encryptor) Option

WithEncryptor enables AES-256-GCM encryption at rest for all stored tokens. Encryption is optional; omit to store tokens in plaintext.

func WithInstrumentation added in v0.2.160 

func WithInstrumentation(inst *instrumentation.Instrumentation) Option

WithInstrumentation wires OpenTelemetry tracing and metrics into the store.

type Store 

type Store struct {
	// contains filtered or unexported fields
}

Store is a Valkey-backed implementation of all storage interfaces. It implements TokenStore, ClientStore, FlowStore, RefreshTokenFamilyStore, and TokenRevocationStore.

func New 

func New(cfg Config, opts ...Option) (*Store, error)

New creates a new Valkey-backed storage instance. All dependencies (encryptor, instrumentation) are supplied at construction via options and are immutable afterward. Returns an error if the connection cannot be established.

func (*Store) AtomicCheckAndMarkAuthCodeUsed 

func (s *Store) AtomicCheckAndMarkAuthCodeUsed(ctx context.Context, code string) (authCode *storage.AuthorizationCode, err error)

AtomicCheckAndMarkAuthCodeUsed atomically checks if a code is unused and marks it as used. This prevents race conditions in authorization code reuse detection. Returns the auth code if successful, or an error if code is already used.

SECURITY: This operation is atomic via Lua script - only ONE concurrent request can succeed.

IMPORTANT: The authCode is ONLY returned on reuse errors (Used=true) to enable detection and revocation. For other errors (not found, expired), nil is returned to prevent information leakage.

func (*Store) AtomicGetAndDeleteRefreshToken 

func (s *Store) AtomicGetAndDeleteRefreshToken(ctx context.Context, refreshToken string) (resUserID, resClientID string, resToken *oauth2.Token, err error)

AtomicGetAndDeleteRefreshToken atomically retrieves and deletes a refresh token. This prevents race conditions in refresh token rotation and reuse detection. Returns the userID, clientID, and provider token if successful.

SECURITY: This operation is atomic via Lua script - only ONE concurrent request can succeed. SECURITY: Returns clientID for client binding validation per OAuth 2.1 Section 6.

func (*Store) CheckIPLimit 

func (s *Store) CheckIPLimit(ctx context.Context, ip string, maxClientsPerIP int) (err error)

CheckIPLimit checks if an IP has reached the client registration limit

func (*Store) Close 

func (s *Store) Close()

Close closes the Valkey client connection.

func (*Store) DeleteAuthorizationCode 

func (s *Store) DeleteAuthorizationCode(ctx context.Context, code string) (err error)

DeleteAuthorizationCode removes an authorization code

func (*Store) DeleteAuthorizationState 

func (s *Store) DeleteAuthorizationState(ctx context.Context, stateID string) (err error)

DeleteAuthorizationState removes an authorization state Removes both client state and provider state entries

func (*Store) DeleteClient added in v0.2.145 

func (s *Store) DeleteClient(ctx context.Context, clientID string) (err error)

DeleteClient removes a registered client by ID.

func (*Store) DeleteRefreshToken 

func (s *Store) DeleteRefreshToken(ctx context.Context, refreshToken string) (err error)

DeleteRefreshToken removes a refresh token

func (*Store) DeleteToken 

func (s *Store) DeleteToken(ctx context.Context, userID string) (err error)

DeleteToken removes a token for a user

func (*Store) GetActiveRefreshTokenByFamily added in v0.2.122 

func (s *Store) GetActiveRefreshTokenByFamily(ctx context.Context, familyID string) (refreshToken, clientID string, err error)

GetActiveRefreshTokenByFamily returns the most recent (highest generation) non-revoked refresh token for the family along with the owning client ID (storage.ActiveRefreshTokenByFamilyStore). Iterates the same Set used by GetRefreshTokenFamilyByID, fetches per-token metadata, and picks the highest-Generation entry whose Revoked flag is unset.

Returns ErrRefreshTokenFamilyNotFound when the Set is empty, or ErrRefreshTokenFamilyRevoked when entries exist but every reachable member's metadata is revoked.

func (*Store) GetAuthorizationCode 

func (s *Store) GetAuthorizationCode(ctx context.Context, code string) (result *storage.AuthorizationCode, err error)

GetAuthorizationCode retrieves an authorization code without modifying it. NOTE: For actual code exchange, use AtomicCheckAndMarkAuthCodeUsed instead to prevent race conditions.

func (*Store) GetAuthorizationState 

func (s *Store) GetAuthorizationState(ctx context.Context, stateID string) (result *storage.AuthorizationState, err error)

GetAuthorizationState retrieves an authorization state by client state

func (*Store) GetAuthorizationStateByProviderState 

func (s *Store) GetAuthorizationStateByProviderState(ctx context.Context, providerState string) (result *storage.AuthorizationState, err error)

GetAuthorizationStateByProviderState retrieves an authorization state by provider state This is used during provider callback validation (separate from client state)

func (*Store) GetClient 

func (s *Store) GetClient(ctx context.Context, clientID string) (result *storage.Client, err error)

GetClient retrieves a client by ID

func (*Store) GetRefreshTokenFamily 

func (s *Store) GetRefreshTokenFamily(ctx context.Context, refreshToken string) (result *storage.RefreshTokenFamilyMetadata, err error)

GetRefreshTokenFamily retrieves family metadata for a refresh token

func (*Store) GetRefreshTokenFamilyByID added in v0.2.120 

func (s *Store) GetRefreshTokenFamilyByID(ctx context.Context, familyID string) (result *storage.RefreshTokenFamilyMetadata, err error)

GetRefreshTokenFamilyByID returns family metadata indexed by family ID (storage.RefreshTokenFamilyByIDStore). Lookups read one member of the {prefix}family:{familyID} Set and follow it to the per-token metadata hash; the family fields (FamilyID, UserID, ClientID, Revoked, RevokedAt, IssuedAt) are identical across the Set's members so any member is representative.

Returns ErrRefreshTokenFamilyNotFound when the Set is empty (also when the family was wiped by retention cleanup).

func (*Store) GetRefreshTokenInfo 

func (s *Store) GetRefreshTokenInfo(ctx context.Context, refreshToken string) (userID string, err error)

GetRefreshTokenInfo retrieves the user ID for a refresh token

func (*Store) GetToken 

func (s *Store) GetToken(ctx context.Context, userID string) (result *oauth2.Token, err error)

GetToken retrieves an oauth2.Token for a user and decrypts if necessary

func (*Store) GetTokenMetadata 

func (s *Store) GetTokenMetadata(tokenID string) (*storage.TokenMetadata, error)

GetTokenMetadata retrieves metadata for a token (including RFC 8707 audience)

func (*Store) GetTokensByUserClient 

func (s *Store) GetTokensByUserClient(ctx context.Context, userID, clientID string) (tokens []string, err error)

GetTokensByUserClient retrieves all token IDs for a user+client combination. Both hashed (current) and legacy (pre-migration) sets are unioned and deduplicated. This is used by Server.RevokeAllTokensForUserClient for provider-side revocation; missing legacy tokens would leave pre-migration tokens unrevoked at the provider.

func (*Store) GetUserInfo 

func (s *Store) GetUserInfo(ctx context.Context, userID string) (result *storage.UserInfo, err error)

GetUserInfo retrieves user information

func (*Store) IsJTIRevoked added in v0.2.120 

func (s *Store) IsJTIRevoked(ctx context.Context, jti string) (revoked bool, err error)

IsJTIRevoked implements storage.RevokedTokenStore via an EXISTS lookup. EXISTS returns 0 for the no-revocation case (including expired entries already evicted by Valkey TTL), so a missing key is safely interpreted as "not revoked".

func (*Store) ListClients 

func (s *Store) ListClients(ctx context.Context) (result []*storage.Client, err error)

ListClients lists all registered clients

func (*Store) LogValue added in v0.2.123 

func (s *Store) LogValue() slog.Value

LogValue implements slog.LogValuer so callers can emit a structured snapshot of the store's posture: 

logger.Info("storage initialized", "store", store)

func (*Store) RevokeAllTokensForUserClient 

func (s *Store) RevokeAllTokensForUserClient(ctx context.Context, userID, clientID string) (count int, err error)

RevokeAllTokensForUserClient revokes all tokens (access + refresh) for a specific user+client combination. This implements the OAuth 2.1 requirement for authorization code reuse detection. Returns the number of tokens revoked and any error encountered.

func (*Store) RevokeJTI added in v0.2.120 

func (s *Store) RevokeJTI(ctx context.Context, jti string, expiresAt time.Time) (err error)

RevokeJTI implements storage.RevokedTokenStore by writing a sentinel value at revokedJTIKey(jti) with EXAT set to the JWT's expiresAt. The value is intentionally minimal — presence is the signal; the JWT itself already carries every other claim. RFC 7009 treats revocation of an already-expired token as a no-op, so we drop those without a round-trip.

func (*Store) RevokeRefreshTokenFamily 

func (s *Store) RevokeRefreshTokenFamily(ctx context.Context, familyID string) (err error)

RevokeRefreshTokenFamily revokes all tokens in a family (for reuse detection) This is called when token reuse is detected (OAuth 2.1 security requirement)

func (*Store) SaveAuthorizationCode 

func (s *Store) SaveAuthorizationCode(ctx context.Context, code *storage.AuthorizationCode) (err error)

SaveAuthorizationCode saves an issued authorization code

func (*Store) SaveAuthorizationState 

func (s *Store) SaveAuthorizationState(ctx context.Context, state *storage.AuthorizationState) (err error)

SaveAuthorizationState saves the state of an ongoing authorization flow Stores by both client state (StateID) and provider state (ProviderState) for dual lookup

func (*Store) SaveClient 

func (s *Store) SaveClient(ctx context.Context, client *storage.Client) (err error)

SaveClient saves a registered client

func (*Store) SaveRefreshToken 

func (s *Store) SaveRefreshToken(ctx context.Context, refreshToken, userID string, expiresAt time.Time) (err error)

SaveRefreshToken saves a refresh token mapping to user ID with expiry

func (*Store) SaveRefreshTokenWithFamily 

func (s *Store) SaveRefreshTokenWithFamily(ctx context.Context, refreshToken, userID, clientID, familyID string, generation int, expiresAt time.Time) (err error)

SaveRefreshTokenWithFamily saves a refresh token with family tracking for reuse detection This is the OAuth 2.1 compliant version that enables token theft detection

func (*Store) SaveToken 

func (s *Store) SaveToken(ctx context.Context, userID string, token *oauth2.Token) (err error)

SaveToken saves an oauth2.Token for a user with optional encryption at rest

func (*Store) SaveTokenMetadata 

func (s *Store) SaveTokenMetadata(ctx context.Context, tokenID string, metadata storage.TokenMetadata) error

SaveTokenMetadata saves metadata for a token. Implements storage.TokenMetadataStore. IssuedAt and ExpiresAt are set by the caller and persisted as-is. When ExpiresAt is non-zero the key TTL is set accordingly so stale metadata is evicted automatically.

func (*Store) SaveUserInfo 

func (s *Store) SaveUserInfo(ctx context.Context, userID string, info *storage.UserInfo) (err error)

SaveUserInfo saves user information

func (*Store) TrackClientIP 

func (s *Store) TrackClientIP(ctx context.Context, _ string, ip string) (err error)

TrackClientIP increments the client count for an IP address. The clientID parameter is accepted for interface compatibility but is not stored separately; only the IP address is tracked.

func (*Store) ValidateClientSecret 

func (s *Store) ValidateClientSecret(ctx context.Context, clientID, clientSecret string) (err error)

ValidateClientSecret validates a client's secret using bcrypt Uses constant-time operations to prevent timing attacks

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.