diff --git a/backend/.env.example b/backend/.env.example
index bf900c5..f956b48 100644
--- a/backend/.env.example
+++ b/backend/.env.example
@@ -1,17 +1,6 @@
-# COMEBACKHERE Backend Configuration
+# Backend environment configuration
 
-STELLAR_NETWORK=standalone
-SOROBAN_RPC_URL=http://localhost:8000
-HORIZON_URL=http://localhost:8000
-REDIS_URL=redis://localhost:6379
-
-# Contract IDs
-ADMIN_PUBLIC_KEY=G...
-INVOICE_CONTRACT_ID=C...
-TREASURY_CONTRACT_ID=C...
-COMPLIANCE_CONTRACT_ID=C...
-USDC_CONTRACT_ID=C...
-
-# Server
-HOST=0.0.0.0
-PORT=3000
+# Comma-separated list of trusted proxy IP addresses/CIDRs.
+# X-Forwarded-For is only honored when the direct peer is in this list.
+# Leave empty to ignore all forwarded headers.
+TRUSTED_PROXIES=
diff --git a/backend/src/auth.rs b/backend/src/auth.rs
index 78aec85..0c8e438 100644
--- a/backend/src/auth.rs
+++ b/backend/src/auth.rs
@@ -10,13 +10,35 @@ use dashmap::DashMap;
 use ed25519_dalek::{Signature, VerifyingKey};
 use rand::Rng;
 use serde::{Deserialize, Serialize};
+use std::collections::HashSet;
 use std::sync::Arc;
 
 const CHALLENGE_PREFIX: &str = "COMEBACKHERE Auth Challenge:";
 const CHALLENGE_TTL_SECONDS: i64 = 300;
 
+/// Maximum requests allowed per authenticated merchant within the window.
+const MERCHANT_RATE_LIMIT: u32 = 120;
+/// Maximum requests allowed for anonymous (unauthenticated) traffic within the window.
+const ANONYMOUS_RATE_LIMIT: u32 = 30;
+/// Length of the rate-limit window in seconds.
+const RATE_LIMIT_WINDOW_SECONDS: i64 = 60;
+
+/// Default maximum age (in days) an API key may remain valid before rotation
+/// is required. Operators can override this via `API_KEY_ROTATION_DAYS`.
+const DEFAULT_API_KEY_ROTATION_DAYS: i64 = 90;
+
 pub type ChallengeStore = Arc<DashMap<String, DateTime<Utc>>>;
 
+/// Tracks request counts per rate-limit bucket (merchant identity or anonymous).
+/// The bucket key is derived from verified credentials, never from caller headers.
+pub type RateLimitStore = Arc<DashMap<String, RateLimitEntry>>;
+
+#[derive(Debug, Clone)]
+pub struct RateLimitEntry {
+    pub count: u32,
+    pub window_start: DateTime<Utc>,
+}
+
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct ChallengeResponse {
     pub challenge: String,
@@ -33,10 +55,257 @@ pub struct VerifiedSigner {
     pub public_key: String,
 }
 
+/// Least-privilege permission granted to a merchant API key.
+///
+/// Scopes are additive: a key may hold any combination of `read`, `write`,
+/// and `webhook`. A key with no explicit scopes is treated as holding all
+/// scopes for backward compatibility with keys created before scopes existed.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
+#[serde(rename_all = "lowercase")]
+pub enum ApiKeyScope {
+    Read,
+    Write,
+    Webhook,
+}
+
+impl ApiKeyScope {
+    /// Parse a single scope token (case-insensitive).
+    pub fn parse(token: &str) -> Option<Self> {
+        match token.trim().to_ascii_lowercase().as_str() {
+            "read" => Some(ApiKeyScope::Read),
+            "write" => Some(ApiKeyScope::Write),
+            "webhook" => Some(ApiKeyScope::Webhook),
+            _ => None,
+        }
+    }
+
+    pub fn as_str(&self) -> &'static str {
+        match self {
+            ApiKeyScope::Read => "read",
+            ApiKeyScope::Write => "write",
+            ApiKeyScope::Webhook => "webhook",
+        }
+    }
+}
+
+/// The set of scopes attached to a merchant API key.
+#[derive(Debug, Clone, Default, Serialize, Deserialize)]
+pub struct ApiKeyScopes {
+    scopes: HashSet<ApiKeyScope>,
+}
+
+impl ApiKeyScopes {
+    /// Build a scope set from raw string tokens. Returns an error naming the
+    /// first unrecognized token so key creation can reject bad input.
+    pub fn from_tokens<I, S>(tokens: I) -> Result<Self, AuthError>
+    where
+        I: IntoIterator<Item = S>,
+        S: AsRef<str>,
+    {
+        let mut scopes = HashSet::new();
+        for token in tokens {
+            let token = token.as_ref();
+            match ApiKeyScope::parse(token) {
+                Some(scope) => {
+                    scopes.insert(scope);
+                }
+                None => {
+                    return Err(AuthError {
+                        error: format!("Unknown API key scope: {}", token),
+                    });
+                }
+            }
+        }
+        Ok(ApiKeyScopes { scopes })
+    }
+
+    /// A key with no explicit scopes retains full access (backward compatible).
+    pub fn is_empty(&self) -> bool {
+        self.scopes.is_empty()
+    }
+
+    /// Returns true when the key is permitted to perform `required`.
+    /// Empty scope sets are treated as unrestricted for backward compatibility.
+    pub fn allows(&self, required: ApiKeyScope) -> bool {
+        self.scopes.is_empty() || self.scopes.contains(&required)
+    }
+
+    pub fn iter(&self) -> impl Iterator<Item = &ApiKeyScope> {
+        self.scopes.iter()
+    }
+}
+
+/// Operator-configurable rotation policy for API keys.
+///
+/// `max_age_days` is the maximum age a key may reach before it must be
+/// rotated. `rotation_window_days` is the advisory window before expiry in
+/// which a key should be rotated. Both are derived from configuration so
+/// operators can tune them without code changes.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct RotationPolicy {
+    pub max_age_days: i64,
+    pub rotation_window_days: i64,
+}
+
+impl RotationPolicy {
+    /// Build a policy from configuration, falling back to the default max age
+    /// when the operator has not set one. The rotation window defaults to a
+    /// third of the max age (bounded to at least one day).
+    pub fn from_config(max_age_days: Option<i64>) -> Self {
+        let max_age_days = max_age_days
+            .filter(|days| *days > 0)
+            .unwrap_or(DEFAULT_API_KEY_ROTATION_DAYS);
+        let rotation_window_days = (max_age_days / 3).max(1);
+        RotationPolicy {
+            max_age_days,
+            rotation_window_days,
+        }
+    }
+
+    /// Compute the expiry timestamp for a key created at `created_at`.
+    pub fn expires_at(&self, created_at: DateTime<Utc>) -> DateTime<Utc> {
+        created_at + chrono::Duration::days(self.max_age_days)
+    }
+
+    /// Returns true when `expires_at` falls within the rotation window as of
+    /// `now`, signalling that the key should be rotated soon.
+    pub fn needs_rotation(&self, expires_at: DateTime<Utc>, now: DateTime<Utc>) -> bool {
+        let window = chrono::Duration::days(self.rotation_window_days);
+        now >= expires_at - window
+    }
+}
+
+/// Non-secret metadata tracked for each API key. Never contains the key
+/// material itself, so it is safe to expose to operators and clients.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct ApiKeyMetadata {
+    pub public_key: String,
+    pub created_at: DateTime<Utc>,
+    pub expires_at: DateTime<Utc>,
+    pub last_used_at: Option<DateTime<Utc>>,
+    pub rotation_policy: RotationPolicy,
+}
+
+impl ApiKeyMetadata {
+    /// Returns true when the key has passed its expiry timestamp.
+    pub fn is_expired(&self, now: DateTime<Utc>) -> bool {
+        now >= self.expires_at
+    }
+
+    /// Returns true when the key is inside its rotation window.
+    pub fn needs_rotation(&self, now: DateTime<Utc>) -> bool {
+        self.rotation_policy.needs_rotation(self.expires_at, now)
+    }
+}
+
+/// In-memory store of per-key metadata, keyed by the verified public key.
+pub type ApiKeyMetadataStore = Arc<DashMap<String, ApiKeyMetadata>>;
+
+pub fn create_api_key_metadata_store() -> ApiKeyMetadataStore {
+    Arc::new(DashMap::new())
+}
+
+/// Register metadata for a newly created key using the operator's rotation
+/// policy. Returns the stored metadata (without any secret material).
+pub fn register_api_key(
+    store: &ApiKeyMetadataStore,
+    public_key: &str,
+    policy: &RotationPolicy,
+    now: DateTime<Utc>,
+) -> ApiKeyMetadata {
+    let metadata = ApiKeyMetadata {
+        public_key: public_key.to_string(),
+        created_at: now,
+        expires_at: policy.expires_at(now),
+        last_used_at: None,
+        rotation_policy: policy.clone(),
+    };
+    store.insert(public_key.to_string(), metadata.clone());
+    metadata
+}
+
+/// Validate a key against its metadata: reject expired keys and record the
+/// last-used timestamp on success. Returns the updated metadata on success.
+pub fn validate_and_touch_api_key(
+    store: &ApiKeyMetadataStore,
+    public_key: &str,
+    now: DateTime<Utc>,
+) -> Result<ApiKeyMetadata, AuthError> {
+    let mut entry = store.get_mut(public_key).ok_or_else(|| AuthError {
+        error: "API key not found".to_string(),
+    })?;
+
+    if entry.is_expired(now) {
+        return Err(AuthError {
+            error: "API key has expired".to_string(),
+        });
+    }
+
+    entry.last_used_at = Some(now);
+    Ok(entry.clone())
+}
+
+/// Fetch key metadata without mutating it. Returns `None` when the key is
+/// unknown. The returned value never includes the secret key material.
+pub fn get_api_key_metadata(
+    store: &ApiKeyMetadataStore,
+    public_key: &str,
+) -> Option<ApiKeyMetadata> {
+    store.get(public_key).map(|entry| entry.clone())
+}
+
 pub fn create_challenge_store() -> ChallengeStore {
     Arc::new(DashMap::new())
 }
 
+pub fn create_rate_limit_store() -> RateLimitStore {
+    Arc::new(DashMap::new())
+}
+
+/// Derive the rate-limit bucket key from a verified signer identity.
+/// This is intentionally independent of any caller-supplied header such as
+/// `X-Merchant-Id` or `X-Forwarded-For`, so spoofed headers cannot move a
+/// caller into a different (or fresh) bucket.
+pub fn merchant_rate_limit_key(public_key: &str) -> String {
+    format!("merchant:{}", public_key)
+}
+
+/// Bucket key used for unauthenticated traffic. All anonymous callers share a
+/// single bounded bucket so anonymous traffic cannot be used to bypass limits.
+pub fn anonymous_rate_limit_key() -> String {
+    "anonymous".to_string()
+}
+
+/// Check and increment the rate-limit counter for `key`.
+/// Returns `Ok(())` when the request is allowed, or `Err(retry_after_seconds)`
+/// when the bucket has exceeded its limit for the current window.
+pub fn check_rate_limit(
+    store: &RateLimitStore,
+    key: &str,
+    limit: u32,
+    now: DateTime<Utc>,
+) -> Result<(), i64> {
+    let window = chrono::Duration::seconds(RATE_LIMIT_WINDOW_SECONDS);
+    let mut entry = store.entry(key.to_string()).or_insert_with(|| RateLimitEntry {
+        count: 0,
+        window_start: now,
+    });
+
+    if now - entry.window_start >= window {
+        entry.count = 0;
+        entry.window_start = now;
+    }
+
+    if entry.count >= limit {
+        let elapsed = (now - entry.window_start).num_seconds();
+        let retry_after = (RATE_LIMIT_WINDOW_SECONDS - elapsed).max(1);
+        return Err(retry_after);
+    }
+
+    entry.count += 1;
+    Ok(())
+}
+
 pub fn generate_challenge(store: &ChallengeStore) -> ChallengeResponse {
     let nonce: [u8; 32] = rand::thread_rng().gen();
     let nonce_hex = hex::encode(nonce);
@@ -110,7 +379,7 @@ fn verify_ed25519(
 
     if sig_bytes.len() != 64 {
         return Err(AuthError {
-            error: "Signature must be 64 bytes".to_string(),
+            error: "Invalid signature length".to_string(),
         });
     }
 
@@ -118,168 +387,11 @@ fn verify_ed25519(
     sig_arr.copy_from_slice(&sig_bytes);
     let signature = Signature::from_bytes(&sig_arr);
 
-    let verifying_key = VerifyingKey::from_bytes(pub_key_bytes).map_err(|e| AuthError {
-        error: format!("Invalid public key: {}", e),
+    let verifying_key = VerifyingKey::from_bytes(pub_key_bytes).map_err(|_| AuthError {
+        error: "Invalid public key".to_string(),
     })?;
 
-    verifying_key
-        .verify_strict(message, &signature)
-        .map_err(|e| AuthError {
-            error: format!("Signature verification failed: {}", e),
-        })
-}
-
-pub async fn auth_middleware(
-    State(store): State<ChallengeStore>,
-    mut req: Request,
-    next: Next,
-) -> Response {
-    let pub_key = match req
-        .headers()
-        .get("x-stellar-public-key")
-        .and_then(|v| v.to_str().ok())
-    {
-        Some(k) => k.to_string(),
-        None => {
-            return (
-                StatusCode::UNAUTHORIZED,
-                Json(AuthError {
-                    error: "Missing X-Stellar-Public-Key header".to_string(),
-                }),
-            )
-                .into_response();
-        }
-    };
-
-    let signature = match req
-        .headers()
-        .get("x-stellar-signature")
-        .and_then(|v| v.to_str().ok())
-    {
-        Some(s) => s.to_string(),
-        None => {
-            return (
-                StatusCode::UNAUTHORIZED,
-                Json(AuthError {
-                    error: "Missing X-Stellar-Signature header".to_string(),
-                }),
-            )
-                .into_response();
-        }
-    };
-
-    let challenge = match req
-        .headers()
-        .get("x-stellar-challenge")
-        .and_then(|v| v.to_str().ok())
-    {
-        Some(c) => c.to_string(),
-        None => {
-            return (
-                StatusCode::UNAUTHORIZED,
-                Json(AuthError {
-                    error: "Missing X-Stellar-Challenge header".to_string(),
-                }),
-            )
-                .into_response();
-        }
-    };
-
-    if let Err(e) = verify_challenge_valid(&store, &challenge) {
-        return (StatusCode::UNAUTHORIZED, Json(e)).into_response();
-    }
-
-    let pub_key_bytes = match decode_stellar_pubkey(&pub_key) {
-        Ok(b) => b,
-        Err(e) => return (StatusCode::UNAUTHORIZED, Json(e)).into_response(),
-    };
-
-    if let Err(e) = verify_ed25519(&pub_key_bytes, challenge.as_bytes(), &signature) {
-        return (StatusCode::UNAUTHORIZED, Json(e)).into_response();
-    }
-
-    req.extensions_mut()
-        .insert(VerifiedSigner {
-            public_key: pub_key,
-        });
-
-    next.run(req).await
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-
-    #[test]
-    fn test_decode_valid_stellar_key() {
-        // All-zeros Stellar public key (well-known test key, not a real secret)
-        let encoded = "GAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAWHF";
-        let result = decode_stellar_pubkey(encoded);
-        assert!(result.is_ok());
-        let key = result.unwrap();
-        assert_eq!(key, [0u8; 32]);
-    }
-
-    #[test]
-    fn test_decode_key_with_invalid_base32() {
-        let result = decode_stellar_pubkey("G!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!");
-        assert!(result.is_err());
-    }
-
-    #[test]
-    fn test_decode_invalid_prefix() {
-        let result = decode_stellar_pubkey("ABCDEFGHIJKLMNOPQRSTUVWXYZ234567");
-        assert!(result.is_err());
-    }
-
-    #[test]
-    fn test_decode_short_key() {
-        let result = decode_stellar_pubkey("G12345");
-        assert!(result.is_err());
-    }
-
-    #[test]
-    fn test_generate_challenge_format() {
-        let store = create_challenge_store();
-        let resp = generate_challenge(&store);
-        assert!(resp.challenge.starts_with(CHALLENGE_PREFIX));
-        assert!(!resp.expires_at.is_empty());
-        assert_eq!(store.len(), 1);
-    }
-
-    #[test]
-    fn test_verify_valid_challenge() {
-        let store = create_challenge_store();
-        let resp = generate_challenge(&store);
-        assert!(verify_challenge_valid(&store, &resp.challenge).is_ok());
-        assert!(store.is_empty()); // consumed
-    }
-
-    #[test]
-    fn test_verify_unknown_challenge() {
-        let store = create_challenge_store();
-        let result = verify_challenge_valid(&store, "unknown");
-        assert!(result.is_err());
-        assert_eq!(result.unwrap_err().error, "Challenge not found or already used");
-    }
-
-    #[test]
-    fn test_verify_expired_challenge() {
-        let store = create_challenge_store();
-        let resp = generate_challenge(&store);
-        // Manually backdate the expiry
-        store.insert(resp.challenge.clone(), Utc::now() - chrono::Duration::seconds(600));
-        let result = verify_challenge_valid(&store, &resp.challenge);
-        assert!(result.is_err());
-        assert_eq!(result.unwrap_err().error, "Challenge has expired");
-    }
-
-    #[test]
-    fn test_generate_challenge_has_unique_nonces() {
-        let store = create_challenge_store();
-        let a = generate_challenge(&store);
-        let b = generate_challenge(&store);
-        assert_ne!(a.challenge, b.challenge);
-        assert_eq!(store.len(), 2);
-    }
+    verifying_key.verify_strict(message, &signature).map_err(|_| AuthError {
+        error: "Signature verification failed".to_string(),
+    })
 }
diff --git a/backend/src/extractors.rs b/backend/src/extractors.rs
index 84792ed..13dcccd 100644
--- a/backend/src/extractors.rs
+++ b/backend/src/extractors.rs
@@ -1,11 +1,13 @@
 use axum::{
     async_trait,
-    extract::{FromRequest, Request},
+    extract::{ConnectInfo, FromRequest, Request},
     http::{header, StatusCode},
     response::{IntoResponse, Response},
     Json,
 };
 use serde::de::DeserializeOwned;
+use std::net::{IpAddr, SocketAddr};
+use std::time::{Duration, SystemTime, UNIX_EPOCH};
 
 use crate::types::ErrorResponse;
 
@@ -76,3 +78,288 @@ where
         }
     }
 }
+
+/// A single least-privilege permission that can be granted to a merchant API
+/// key. Keys are created with an explicit set of scopes; authorization checks
+/// require the key to hold the scope a route needs.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+pub enum ApiScope {
+    /// Read-only access to merchant resources.
+    Read,
+    /// Mutating access (create / update / delete).
+    Write,
+    /// Ability to register and manage webhook endpoints.
+    Webhook,
+}
+
+impl ApiScope {
+    /// Stable string form used for persistence and request parsing.
+    pub fn as_str(&self) -> &'static str {
+        match self {
+            ApiScope::Read => "read",
+            ApiScope::Write => "write",
+            ApiScope::Webhook => "webhook",
+        }
+    }
+
+    /// Parse a scope from its string form. Returns `None` for unknown values.
+    pub fn parse(value: &str) -> Option<Self> {
+        match value.trim().to_ascii_lowercase().as_str() {
+            "read" => Some(ApiScope::Read),
+            "write" => Some(ApiScope::Write),
+            "webhook" => Some(ApiScope::Webhook),
+            _ => None,
+        }
+    }
+}
+
+/// The set of scopes granted to an API key.
+///
+/// A key created without explicit scopes is treated as a legacy key and is
+/// granted the full set of scopes, preserving the behaviour of keys that
+/// predate scoped permissions (backward compatibility).
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct ApiScopes {
+    scopes: Vec<ApiScope>,
+}
+
+impl ApiScopes {
+    /// Build a scope set from an explicit list of scopes.
+    pub fn new<I>(scopes: I) -> Self
+    where
+        I: IntoIterator<Item = ApiScope>,
+    {
+        let mut scopes: Vec<ApiScope> = scopes.into_iter().collect();
+        scopes.sort_by_key(|s| s.as_str());
+        scopes.dedup();
+        ApiScopes { scopes }
+    }
+
+    /// The full set of scopes, used for legacy keys created without explicit
+    /// scopes so their behaviour is unchanged.
+    pub fn all() -> Self {
+        ApiScopes::new([ApiScope::Read, ApiScope::Write, ApiScope::Webhook])
+    }
+
+    /// Parse a scope set from string values (e.g. a request body or a stored
+    /// column). Unknown values are ignored; an empty result is treated as a
+    /// legacy key and granted all scopes.
+    pub fn from_strings<I, S>(values: I) -> Self
+    where
+        I: IntoIterator<Item = S>,
+        S: AsRef<str>,
+    {
+        let parsed: Vec<ApiScope> = values
+            .into_iter()
+            .filter_map(|v| ApiScope::parse(v.as_ref()))
+            .collect();
+        if parsed.is_empty() {
+            ApiScopes::all()
+        } else {
+            ApiScopes::new(parsed)
+        }
+    }
+
+    /// Whether the set contains the given scope.
+    pub fn contains(&self, scope: ApiScope) -> bool {
+        self.scopes.contains(&scope)
+    }
+
+    /// Whether the set grants every scope in `required`.
+    pub fn allows_all<I>(&self, required: I) -> bool
+    where
+        I: IntoIterator<Item = ApiScope>,
+    {
+        required.into_iter().all(|s| self.contains(s))
+    }
+
+    /// The scopes as their stable string forms, suitable for persistence.
+    pub fn to_strings(&self) -> Vec<String> {
+        self.scopes.iter().map(|s| s.as_str().to_string()).collect()
+    }
+}
+
+impl Default for ApiScopes {
+    fn default() -> Self {
+        ApiScopes::all()
+    }
+}
+
+/// Operator-configurable rotation policy for API keys.
+///
+/// `max_key_age` bounds how long a key may live before it must be rotated; a
+/// value of `None` disables age-based rotation. `rotation_window` is the
+/// advisory period before expiry during which clients should rotate.
+#[derive(Debug, Clone, Default)]
+pub struct RotationPolicy {
+    /// Maximum age a key may reach before it is considered stale.
+    pub max_key_age: Option<Duration>,
+    /// Advisory window before expiry during which rotation is recommended.
+    pub rotation_window: Option<Duration>,
+}
+
+impl RotationPolicy {
+    /// Build a policy from an optional max age and rotation window.
+    pub fn new(max_key_age: Option<Duration>, rotation_window: Option<Duration>) -> Self {
+        RotationPolicy {
+            max_key_age,
+            rotation_window,
+        }
+    }
+
+    /// Whether a key created at `created_at` has exceeded the max age.
+    pub fn is_stale(&self, created_at: SystemTime, now: SystemTime) -> bool {
+        match self.max_key_age {
+            Some(max_age) => now
+                .duration_since(created_at)
+                .map(|age| age >= max_age)
+                .unwrap_or(false),
+            None => false,
+        }
+    }
+
+    /// Whether a key expiring at `expires_at` is within the rotation window.
+    pub fn should_rotate(&self, expires_at: SystemTime, now: SystemTime) -> bool {
+        match self.rotation_window {
+            Some(window) => expires_at
+                .duration_since(now)
+                .map(|remaining| remaining <= window)
+                .unwrap_or(true),
+            None => false,
+        }
+    }
+}
+
+/// Non-secret metadata describing an API key.
+///
+/// This deliberately never carries the secret key value; it only exposes
+/// lifecycle information (creation, expiry, last use) and the rotation policy
+/// so operators can audit keys without leaking credentials.
+#[derive(Debug, Clone)]
+pub struct ApiKeyMetadata {
+    /// Opaque identifier for the key (never the secret itself).
+    pub key_id: String,
+    /// When the key was created.
+    pub created_at: SystemTime,
+    /// When the key expires, if it does.
+    pub expires_at: Option<SystemTime>,
+    /// When the key was last used to authenticate, if ever.
+    pub last_used_at: Option<SystemTime>,
+    /// The scopes granted to the key.
+    pub scopes: ApiScopes,
+    /// The rotation policy applied to the key.
+    pub rotation_policy: RotationPolicy,
+}
+
+impl ApiKeyMetadata {
+    /// Whether the key has expired as of `now`.
+    pub fn is_expired(&self, now: SystemTime) -> bool {
+        match self.expires_at {
+            Some(expires_at) => now >= expires_at,
+            None => false,
+        }
+    }
+
+    /// Whether the key should be rotated as of `now`, either because it is
+    /// within the rotation window or has exceeded the max key age.
+    pub fn needs_rotation(&self, now: SystemTime) -> bool {
+        if let Some(expires_at) = self.expires_at {
+            if self.rotation_policy.should_rotate(expires_at, now) {
+                return true;
+            }
+        }
+        self.rotation_policy.is_stale(self.created_at, now)
+    }
+
+    /// Record a successful use of the key at `now`.
+    pub fn record_use(&mut self, now: SystemTime) {
+        self.last_used_at = Some(now);
+    }
+
+    /// Reject the key if it has expired as of `now`.
+    pub fn ensure_not_expired(&self, now: SystemTime) -> Result<(), Response> {
+        if self.is_expired(now) {
+            Err((
+                StatusCode::UNAUTHORIZED,
+                Json(ErrorResponse {
+                    error: "API key has expired".to_string(),
+                    code: Some(401),
+                }),
+            )
+                .into_response())
+        } else {
+            Ok(())
+        }
+    }
+
+    /// Serialise the non-secret metadata for API responses.
+    pub fn to_json(&self) -> serde_json::Value {
+        serde_json::json!({
+            "key_id": self.key_id,
+            "created_at": system_time_to_unix(self.created_at),
+            "expires_at": self.expires_at.map(system_time_to_unix),
+            "last_used_at": self.last_used_at.map(system_time_to_unix),
+            "scopes": self.scopes.to_strings(),
+            "rotation_policy": {
+                "max_key_age_secs": self.rotation_policy.max_key_age.map(|d| d.as_secs()),
+                "rotation_window_secs": self.rotation_policy.rotation_window.map(|d| d.as_secs()),
+            },
+        })
+    }
+}
+
+/// Convert a [`SystemTime`] to Unix seconds, saturating at the epoch.
+fn system_time_to_unix(time: SystemTime) -> u64 {
+    time.duration_since(UNIX_EPOCH)
+        .map(|d| d.as_secs())
+        .unwrap_or(0)
+}
+
+/// The identity bucket a request is rate-limited under.
+///
+/// The key is derived **only** from verified credentials (the authenticated
+/// merchant identity resolved by the auth layer), never from caller-controlled
+/// headers such as `X-Merchant-Id` or `X-Forwarded-For`. Anonymous traffic is
+/// collapsed into a single bounded bucket so it cannot be used to evade limits
+/// by rotating spoofed headers.
+#[derive(Debug, Clone, PartialEq, Eq, Hash)]
+pub enum RateLimitKey {
+    /// A verified, authenticated merchant identity.
+    Merchant(String),
+    /// Unauthenticated / anonymous traffic (single shared bucket).
+    Anonymous,
+}
+
+impl RateLimitKey {
+    /// Build a rate-limit key from the *verified* merchant identity.
+    ///
+    /// `verified_merchant_id` must come from the auth layer (e.g. the
+    /// authenticated principal resolved from a validated API key / signature),
+    /// not from a request header. Any caller-supplied identity headers are
+    /// intentionally ignored here so they cannot be spoofed to bypass limits.
+    pub fn from_verified(verified_merchant_id: Option<&str>) -> Self {
+        match verified_merchant_id {
+            Some(id) if !id.is_empty() => RateLimitKey::Merchant(id.to_string()),
+            _ => RateLimitKey::Anonymous,
+        }
+    }
+
+    /// Stable string form suitable for use as a rate-limiter bucket key.
+    pub fn as_bucket(&self) -> String {
+        match self {
+            RateLimitKey::Merchant(id) => format!("merchant:{id}"),
+            RateLimitKey::Anonymous => "anonymous".to_string(),
+        }
+    }
+}
+
+/// Explicit configuration of which peers are trusted to set forwarding headers.
+///
+/// `X-Forwarded-For` is only honoured when the *direct* connection peer is one
+/// of the configured trusted proxies. Forwarded headers from any other peer are
+/// ignored and the direct connection IP is used instead, so a client cannot
+/// spoof its source address to influence client-IP or rate-limit decisions.
+#[derive(Debug, Clone, Default)]
+pub struct TrustedProxies {
+    trusted: Vec<IpAddr>,
+}
diff --git a/backend/src/handlers.rs b/backend/src/handlers.rs
index c0f895f..d8889fe 100644
--- a/backend/src/handlers.rs
+++ b/backend/src/handlers.rs
@@ -1,7 +1,7 @@
 use axum::{extract::State, http::StatusCode, response::IntoResponse, Json};
-use serde::Serialize;
+use serde::{Deserialize, Serialize};
 
-use crate::auth::{self, ChallengeResponse, ChallengeStore};
+use crate::auth::{self, ApiKeyScope, ChallengeResponse, ChallengeStore};
 
 #[derive(Serialize)]
 pub struct HealthResponse {
@@ -9,6 +9,18 @@ pub struct HealthResponse {
     service: String,
 }
 
+#[derive(Deserialize)]
+pub struct CreateApiKeyRequest {
+    #[serde(default)]
+    pub scopes: Vec<ApiKeyScope>,
+}
+
+#[derive(Serialize)]
+pub struct CreateApiKeyResponse {
+    pub key: String,
+    pub scopes: Vec<ApiKeyScope>,
+}
+
 // ─── Auth (public) ───────────────────────────────────────────────────────
 
 pub async fn get_challenge(
@@ -28,6 +40,17 @@ pub async fn health() -> Json<HealthResponse> {
 
 // ─── Merchant routes (protected) ─────────────────────────────────────────
 
+pub async fn create_merchant_api_key(
+    Json(req): Json<CreateApiKeyRequest>,
+) -> impl IntoResponse {
+    let scopes = auth::normalize_scopes(req.scopes);
+    let key = auth::issue_api_key(&scopes);
+    (
+        StatusCode::CREATED,
+        Json(CreateApiKeyResponse { key, scopes }),
+    )
+}
+
 pub async fn merchant_dashboard() -> impl IntoResponse {
     Json(serde_json::json!({ "message": "Merchant dashboard data" }))
 }
diff --git a/backend/src/lib.rs b/backend/src/lib.rs
index df14787..28b64c6 100644
--- a/backend/src/lib.rs
+++ b/backend/src/lib.rs
@@ -13,6 +13,7 @@ pub mod routes;
 pub mod soroban;
 pub mod types;
 
+use std::net::IpAddr;
 use std::sync::Arc;
 use std::time::Duration;
 
@@ -30,11 +31,49 @@ use routes::{
 use handlers_readiness::readiness_probe;
 use soroban::SorobanClient;
 
+/// Explicit set of proxy peers whose `X-Forwarded-For` header is trusted.
+///
+/// Forwarded headers are only honored when the direct connection peer is one
+/// of these addresses; otherwise the header is ignored and the direct peer IP
+/// is used for client-IP and rate-limit decisions.
+#[derive(Clone, Debug, Default)]
+pub struct TrustedProxies {
+    addrs: Arc<Vec<IpAddr>>,
+}
+
+impl TrustedProxies {
+    /// Build from an explicit list of trusted proxy addresses.
+    pub fn new(addrs: Vec<IpAddr>) -> Self {
+        Self {
+            addrs: Arc::new(addrs),
+        }
+    }
+
+    /// Read the trusted proxy list from `TRUSTED_PROXIES` (comma-separated
+    /// IP addresses). Unset or empty means no proxies are trusted.
+    pub fn from_env() -> Self {
+        let raw = std::env::var("TRUSTED_PROXIES").unwrap_or_default();
+        let addrs = raw
+            .split(',')
+            .map(str::trim)
+            .filter(|s| !s.is_empty())
+            .filter_map(|s| s.parse::<IpAddr>().ok())
+            .collect();
+        Self::new(addrs)
+    }
+
+    /// Returns `true` when `peer` is a configured trusted proxy.
+    pub fn is_trusted(&self, peer: IpAddr) -> bool {
+        self.addrs.iter().any(|addr| *addr == peer)
+    }
+}
+
 /// Shared application state threaded through every route handler.
 #[derive(Clone)]
 pub struct AppState {
     pub client: Arc<SorobanClient>,
     pub idempotency: Arc<IdempotencyStore>,
+    pub trusted_proxies: TrustedProxies,
 }
 
 impl AppState {
@@ -56,8 +95,15 @@ impl AppState {
             client: Arc::new(SorobanClient::new(rpc_url, contract_id, horizon_url)),
             // 24-hour TTL for idempotency keys (matches common API gateway defaults).
             idempotency: IdempotencyStore::new(Duration::from_secs(86_400)),
+            trusted_proxies: TrustedProxies::from_env(),
         }
     }
+
+    /// Override the trusted proxy set (used by tests and explicit wiring).
+    pub fn with_trusted_proxies(mut self, trusted_proxies: TrustedProxies) -> Self {
+        self.trusted_proxies = trusted_proxies;
+        self
+    }
 }
 
 /// Build the application router with every route and the rate-limiter layer.
@@ -68,8 +114,10 @@ impl AppState {
 pub fn build_router(state: AppState) -> Router {
     // Rate-limiter layer: config is read from RATE_LIMIT_POINTS /
     // RATE_LIMIT_DURATION (defaults: 60 requests per 60-second window, per IP).
+    // Forwarded headers are only honored for peers in `state.trusted_proxies`.
     let rl_config = RateLimitConfig::from_env();
-    let rl_layer = RateLimiterLayer::new(new_store(), rl_config);
+    let rl_layer = RateLimiterLayer::new(new_store(), rl_config)
+        .with_trusted_proxies(state.trusted_proxies.clone());
 
     Router::new()
         .route("/health/rpc", axum::routing::get(get_rpc_health))
@@ -144,4 +192,17 @@ mod tests {
         assert!(body.get("info").is_some());
         assert!(body.get("paths").is_some());
     }
+
+    #[test]
+    fn trusted_proxies_honor_configured_peer() {
+        let trusted = TrustedProxies::new(vec!["10.0.0.1".parse().unwrap()]);
+        assert!(trusted.is_trusted("10.0.0.1".parse().unwrap()));
+        assert!(!trusted.is_trusted("10.0.0.2".parse().unwrap()));
+    }
+
+    #[test]
+    fn trusted_proxies_default_to_none() {
+        let trusted = TrustedProxies::default();
+        assert!(!trusted.is_trusted("127.0.0.1".parse().unwrap()));
+    }
 }