The Secure API Vault System is a critical component of NIJA’s multi-user platform that manages cryptocurrency exchange API credentials for all users. This system ensures that sensitive API keys, secrets, and other credentials are stored, accessed, and rotated securely using industry best practices.
┌─────────────────────────────────────────────────────────────┐
│ User Application Layer │
│ Mobile App / Web Dashboard / Trading Service │
└───────────────────────┬─────────────────────────────────────┘
│ HTTPS/TLS 1.3
▼
┌─────────────────────────────────────────────────────────────┐
│ Vault Gateway API │
│ - JWT authentication │
│ - Request validation │
│ - Rate limiting │
│ - Audit logging │
└───────────────────────┬─────────────────────────────────────┘
│
┌──────────────┼──────────────┬──────────────┐
│ │ │ │
▼ ▼ ▼ ▼
┌──────────────┐ ┌──────────┐ ┌────────────┐ ┌──────────────┐
│ Store │ │ Retrieve│ │ Rotate │ │ Revoke │
│ Service │ │ Service │ │ Service │ │ Service │
└──────┬───────┘ └────┬─────┘ └──────┬─────┘ └──────┬───────┘
│ │ │ │
└──────────────┴──────────────┴──────────────┘
│
▼
┌─────────────────────────────────────────────────────────────┐
│ HashiCorp Vault Core Engine │
│ │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │
│ │ KV Secrets │ │ Transit │ │ AppRole │ │
│ │ Engine │ │ Engine │ │ Auth │ │
│ └──────────────┘ └──────────────┘ └──────────────┘ │
│ │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │
│ │ Policies │ │ Audit Log │ │ Leasing │ │
│ │ Engine │ │ Engine │ │ System │ │
│ └──────────────┘ └──────────────┘ └──────────────┘ │
└─────────────────────────┬───────────────────────────────────┘
│
┌────────────────┼────────────────┐
│ │ │
▼ ▼ ▼
┌──────────────┐ ┌──────────────┐ ┌──────────────┐
│ Encrypted │ │ Unseal Keys │ │ Audit Logs │
│ Storage │ │ (HSM/Cloud) │ │ (S3/GCS) │
│ (PostgreSQL)│ │ │ │ │
└──────────────┘ └──────────────┘ └──────────────┘
Why Vault?
Vault Configuration:
# vault-config.hcl
storage "postgresql" {
connection_url = "postgres://vault:password@postgres:5432/vault?sslmode=require"
ha_enabled = true
max_parallel = 128
}
listener "tcp" {
address = "0.0.0.0:8200"
tls_cert_file = "/vault/tls/vault.crt"
tls_key_file = "/vault/tls/vault.key"
}
seal "awskms" {
region = "us-east-1"
kms_key_id = "alias/vault-unseal-key"
}
ui = true
api_addr = "https://vault.nija.io"
cluster_addr = "https://vault.nija.io:8201"
Secrets Engines:
secret/users/{user_id}/{broker}transit/keys/user-credentialsPython-based FastAPI service that provides a simplified, secure interface to Vault.
Implementation:
# vault_gateway.py
from fastapi import FastAPI, Depends, HTTPException, Security
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
import hvac
from pydantic import BaseModel
from typing import Dict, Optional, List
import logging
logger = logging.getLogger(__name__)
app = FastAPI(title="NIJA Vault Gateway API")
security = HTTPBearer()
# Vault client
vault_client = hvac.Client(
url='https://vault.nija.io:8200',
token=os.getenv('VAULT_TOKEN') # AppRole auth in production
)
class CredentialStore(BaseModel):
"""Request to store user credentials."""
user_id: str
broker: str
api_key: str
api_secret: str
additional_params: Optional[Dict[str, str]] = None
class CredentialRetrieve(BaseModel):
"""Request to retrieve user credentials."""
user_id: str
broker: str
class CredentialRotate(BaseModel):
"""Request to rotate user credentials."""
user_id: str
broker: str
new_api_key: str
new_api_secret: str
# Authentication
async def verify_jwt(credentials: HTTPAuthorizationCredentials = Security(security)):
"""Verify JWT token and extract user info."""
token = credentials.credentials
# Implement JWT verification here
# Return user_id from token
return {"user_id": "user_from_token"}
# Store credentials
@app.post("/v1/credentials/store")
async def store_credentials(
request: CredentialStore,
user: dict = Depends(verify_jwt)
):
"""
Store user credentials in Vault.
Security:
- User can only store their own credentials
- Credentials encrypted at rest by Vault
- Audit log entry created
"""
# Verify user can only store their own credentials
if request.user_id != user["user_id"]:
raise HTTPException(status_code=403, detail="Cannot store credentials for other users")
try:
# Build secret path
secret_path = f"secret/users/{request.user_id}/{request.broker}"
# Prepare secret data
secret_data = {
"api_key": request.api_key,
"api_secret": request.api_secret,
"broker": request.broker,
"created_at": datetime.utcnow().isoformat(),
"created_by": user["user_id"]
}
if request.additional_params:
secret_data["additional_params"] = request.additional_params
# Store in Vault KV v2
vault_client.secrets.kv.v2.create_or_update_secret(
path=secret_path,
secret=secret_data
)
logger.info(f"Stored credentials for user {request.user_id} on {request.broker}")
return {
"status": "success",
"message": "Credentials stored successfully",
"user_id": request.user_id,
"broker": request.broker
}
except Exception as e:
logger.error(f"Failed to store credentials: {e}")
raise HTTPException(status_code=500, detail="Failed to store credentials")
# Retrieve credentials
@app.post("/v1/credentials/retrieve")
async def retrieve_credentials(
request: CredentialRetrieve,
user: dict = Depends(verify_jwt)
):
"""
Retrieve user credentials from Vault.
Security:
- User can only retrieve their own credentials
- Audit log entry created
- Credentials decrypted by Vault
"""
# Verify user can only retrieve their own credentials
if request.user_id != user["user_id"]:
raise HTTPException(status_code=403, detail="Cannot retrieve credentials for other users")
try:
# Build secret path
secret_path = f"secret/users/{request.user_id}/{request.broker}"
# Retrieve from Vault
response = vault_client.secrets.kv.v2.read_secret_version(
path=secret_path
)
secret_data = response['data']['data']
logger.info(f"Retrieved credentials for user {request.user_id} on {request.broker}")
return {
"status": "success",
"credentials": {
"api_key": secret_data["api_key"],
"api_secret": secret_data["api_secret"],
"broker": secret_data["broker"],
"additional_params": secret_data.get("additional_params")
}
}
except hvac.exceptions.InvalidPath:
raise HTTPException(status_code=404, detail="Credentials not found")
except Exception as e:
logger.error(f"Failed to retrieve credentials: {e}")
raise HTTPException(status_code=500, detail="Failed to retrieve credentials")
# Rotate credentials
@app.post("/v1/credentials/rotate")
async def rotate_credentials(
request: CredentialRotate,
user: dict = Depends(verify_jwt)
):
"""
Rotate user credentials (update to new keys).
Security:
- User can only rotate their own credentials
- Old version retained (can rollback)
- Audit log entry created
"""
# Verify user can only rotate their own credentials
if request.user_id != user["user_id"]:
raise HTTPException(status_code=403, detail="Cannot rotate credentials for other users")
try:
# Build secret path
secret_path = f"secret/users/{request.user_id}/{request.broker}"
# Prepare new secret data
secret_data = {
"api_key": request.new_api_key,
"api_secret": request.new_api_secret,
"broker": request.broker,
"rotated_at": datetime.utcnow().isoformat(),
"rotated_by": user["user_id"]
}
# Update in Vault (creates new version)
vault_client.secrets.kv.v2.create_or_update_secret(
path=secret_path,
secret=secret_data
)
logger.info(f"Rotated credentials for user {request.user_id} on {request.broker}")
return {
"status": "success",
"message": "Credentials rotated successfully"
}
except Exception as e:
logger.error(f"Failed to rotate credentials: {e}")
raise HTTPException(status_code=500, detail="Failed to rotate credentials")
# Revoke credentials
@app.delete("/v1/credentials/revoke")
async def revoke_credentials(
user_id: str,
broker: str,
user: dict = Depends(verify_jwt)
):
"""
Revoke (delete) user credentials.
Security:
- User can only revoke their own credentials
- Soft delete (can be recovered from versions)
- Audit log entry created
"""
# Verify user can only revoke their own credentials
if user_id != user["user_id"]:
raise HTTPException(status_code=403, detail="Cannot revoke credentials for other users")
try:
# Build secret path
secret_path = f"secret/users/{user_id}/{broker}"
# Delete from Vault (soft delete - versions retained)
vault_client.secrets.kv.v2.delete_latest_version_of_secret(
path=secret_path
)
logger.info(f"Revoked credentials for user {user_id} on {broker}")
return {
"status": "success",
"message": "Credentials revoked successfully"
}
except Exception as e:
logger.error(f"Failed to revoke credentials: {e}")
raise HTTPException(status_code=500, detail="Failed to revoke credentials")
# List user's brokers
@app.get("/v1/credentials/list")
async def list_user_brokers(
user_id: str,
user: dict = Depends(verify_jwt)
):
"""
List all brokers configured for a user.
Security:
- User can only list their own brokers
- Returns broker names only (no credentials)
"""
# Verify user can only list their own brokers
if user_id != user["user_id"]:
raise HTTPException(status_code=403, detail="Cannot list brokers for other users")
try:
# List secrets under user path
secret_path = f"secret/users/{user_id}"
response = vault_client.secrets.kv.v2.list_secrets(
path=secret_path
)
brokers = response['data']['keys']
return {
"status": "success",
"user_id": user_id,
"brokers": brokers
}
except hvac.exceptions.InvalidPath:
return {
"status": "success",
"user_id": user_id,
"brokers": []
}
except Exception as e:
logger.error(f"Failed to list brokers: {e}")
raise HTTPException(status_code=500, detail="Failed to list brokers")
User Policy (read/write own credentials only):
# user-policy.hcl
path "secret/data/users//*" {
capabilities = ["create", "read", "update", "delete", "list"]
}
path "secret/metadata/users//*" {
capabilities = ["list", "read", "delete"]
}
# Deny access to other users' secrets
path "secret/data/users/*" {
capabilities = ["deny"]
}
Service Policy (for trading engine):
# service-policy.hcl
path "secret/data/users/+/+" {
capabilities = ["read"]
}
path "transit/encrypt/user-credentials" {
capabilities = ["update"]
}
path "transit/decrypt/user-credentials" {
capabilities = ["update"]
}
Admin Policy (for platform administration):
# admin-policy.hcl
path "secret/*" {
capabilities = ["create", "read", "update", "delete", "list"]
}
path "sys/policies/*" {
capabilities = ["create", "read", "update", "delete", "list"]
}
path "sys/mounts/*" {
capabilities = ["create", "read", "update", "delete", "list"]
}
path "auth/*" {
capabilities = ["create", "read", "update", "delete", "list"]
}
AppRole Authentication (for services):
# Service authenticates with AppRole
import hvac
client = hvac.Client(url='https://vault.nija.io:8200')
# Login with AppRole
role_id = os.getenv('VAULT_ROLE_ID')
secret_id = os.getenv('VAULT_SECRET_ID')
response = client.auth.approle.login(
role_id=role_id,
secret_id=secret_id
)
client.token = response['auth']['client_token']
JWT Authentication (for users):
# User authenticates with JWT
jwt_token = "user_jwt_token_here"
response = client.auth.jwt.login(
role='user-role',
jwt=jwt_token
)
client.token = response['auth']['client_token']
Audit Backend Configuration:
# Enable file audit backend
vault audit enable file file_path=/vault/logs/audit.log
# Enable syslog audit backend (optional)
vault audit enable syslog tag="vault" facility="AUTH"
Audit Log Format:
{
"time": "2026-01-27T19:48:59.123456Z",
"type": "request",
"auth": {
"client_token": "hmac-sha256:...",
"accessor": "hmac-sha256:...",
"display_name": "user123",
"policies": ["user-policy"],
"token_policies": ["user-policy"],
"metadata": {
"user_id": "user123"
}
},
"request": {
"id": "abc-123-def-456",
"operation": "read",
"client_token": "hmac-sha256:...",
"client_token_accessor": "hmac-sha256:...",
"namespace": {
"id": "root"
},
"path": "secret/data/users/user123/coinbase",
"data": null,
"remote_address": "203.0.113.1"
},
"response": {
"mount_type": "kv"
}
}
Multi-Node Vault Cluster:
# docker-compose.yml for HA Vault
version: '3.8'
services:
vault-1:
image: vault:1.15
environment:
VAULT_ADDR: http://0.0.0.0:8200
VAULT_API_ADDR: https://vault-1.nija.io:8200
VAULT_CLUSTER_ADDR: https://vault-1.nija.io:8201
volumes:
- ./vault-config.hcl:/vault/config/vault.hcl
- vault-1-data:/vault/data
ports:
- "8200:8200"
- "8201:8201"
command: server
vault-2:
image: vault:1.15
environment:
VAULT_ADDR: http://0.0.0.0:8200
VAULT_API_ADDR: https://vault-2.nija.io:8200
VAULT_CLUSTER_ADDR: https://vault-2.nija.io:8201
volumes:
- ./vault-config.hcl:/vault/config/vault.hcl
- vault-2-data:/vault/data
ports:
- "8210:8200"
- "8211:8201"
command: server
vault-3:
image: vault:1.15
environment:
VAULT_ADDR: http://0.0.0.0:8200
VAULT_API_ADDR: https://vault-3.nija.io:8200
VAULT_CLUSTER_ADDR: https://vault-3.nija.io:8201
volumes:
- ./vault-config.hcl:/vault/config/vault.hcl
- vault-3-data:/vault/data
ports:
- "8220:8200"
- "8221:8201"
command: server
volumes:
vault-1-data:
vault-2-data:
vault-3-data:
Automated Backup Script:
#!/bin/bash
# vault-backup.sh
VAULT_ADDR="https://vault.nija.io:8200"
VAULT_TOKEN="$VAULT_ROOT_TOKEN"
BACKUP_DIR="/vault/backups"
DATE=$(date +%Y%m%d_%H%M%S)
# Create snapshot
vault operator raft snapshot save "${BACKUP_DIR}/vault-snapshot-${DATE}.snap"
# Encrypt snapshot
openssl enc -aes-256-cbc -salt -in "${BACKUP_DIR}/vault-snapshot-${DATE}.snap" \
-out "${BACKUP_DIR}/vault-snapshot-${DATE}.snap.enc" -k "$BACKUP_ENCRYPTION_KEY"
# Upload to S3
aws s3 cp "${BACKUP_DIR}/vault-snapshot-${DATE}.snap.enc" \
"s3://nija-vault-backups/snapshots/" \
--sse AES256
# Cleanup old local snapshots (keep last 7 days)
find "${BACKUP_DIR}" -name "vault-snapshot-*.snap*" -mtime +7 -delete
echo "Backup completed: vault-snapshot-${DATE}.snap.enc"
Recovery Procedure:
#!/bin/bash
# vault-restore.sh
BACKUP_FILE="$1"
VAULT_ADDR="https://vault.nija.io:8200"
# Download from S3
aws s3 cp "s3://nija-vault-backups/snapshots/${BACKUP_FILE}" .
# Decrypt
openssl enc -aes-256-cbc -d -in "${BACKUP_FILE}" \
-out "vault-snapshot.snap" -k "$BACKUP_ENCRYPTION_KEY"
# Restore snapshot
vault operator raft snapshot restore vault-snapshot.snap
echo "Restore completed from: ${BACKUP_FILE}"
Use Auto-Unseal with Cloud KMS:
# AWS KMS Auto-Unseal
seal "awskms" {
region = "us-east-1"
kms_key_id = "alias/vault-unseal-key"
}
# GCP Cloud KMS Auto-Unseal
seal "gcpckms" {
project = "nija-platform"
region = "us-east1"
key_ring = "vault"
crypto_key = "vault-unseal-key"
}
Shamir Secret Sharing (manual unseal):
Revoke root token after initialization:
# Revoke root token
vault token revoke <root_token>
# Generate new root token only when needed (emergency)
vault operator generate-root -init
Use short-lived tokens for admin tasks:
# Create admin token with 1-hour TTL
vault token create -policy=admin-policy -ttl=1h
TLS Configuration:
listener "tcp" {
address = "0.0.0.0:8200"
tls_cert_file = "/vault/tls/vault.crt"
tls_key_file = "/vault/tls/vault.key"
tls_min_version = "tls13"
tls_cipher_suites = [
"TLS_AES_256_GCM_SHA384",
"TLS_AES_128_GCM_SHA256",
"TLS_CHACHA20_POLY1305_SHA256"
]
}
Firewall Rules:
Principle of Least Privilege:
Token Renewal:
# Auto-renew tokens before expiry
def renew_token(client):
while True:
# Renew token
client.auth.token.renew_self()
# Sleep for 80% of TTL
lookup = client.auth.token.lookup_self()
ttl = lookup['data']['ttl']
sleep_time = int(ttl * 0.8)
time.sleep(sleep_time)
Prometheus Metrics:
# vault-exporter config
global:
scrape_interval: 15s
scrape_configs:
- job_name: 'vault'
metrics_path: /v1/sys/metrics
params:
format: ['prometheus']
static_configs:
- targets: ['vault.nija.io:8200']
Critical Alerts:
# Connection pool for Vault client
from hvac import Client
from hvac.adapters import HTTPAdapter
from requests.adapters import HTTPAdapter as RequestsHTTPAdapter
class VaultConnectionPool:
def __init__(self, url, max_connections=100):
adapter = RequestsHTTPAdapter(
pool_connections=max_connections,
pool_maxsize=max_connections,
max_retries=3
)
self.client = Client(url=url)
self.client.session.mount('https://', adapter)
def get_client(self):
return self.client
# Global pool
vault_pool = VaultConnectionPool('https://vault.nija.io:8200')
Cache Credentials in Redis:
import redis
import json
from datetime import timedelta
# Redis client
redis_client = redis.Redis(host='localhost', port=6379, db=0)
def get_credentials_cached(user_id: str, broker: str):
"""
Get credentials with Redis caching.
TTL: 5 minutes (reduce Vault load)
"""
# Check cache first
cache_key = f"creds:{user_id}:{broker}"
cached = redis_client.get(cache_key)
if cached:
return json.loads(cached)
# Fetch from Vault
creds = retrieve_credentials(user_id, broker)
# Cache for 5 minutes
redis_client.setex(
cache_key,
timedelta(minutes=5),
json.dumps(creds)
)
return creds
Cache Invalidation:
async def retrieve_multiple_credentials(requests: List[CredentialRetrieve]):
"""Retrieve multiple credentials in parallel."""
tasks = [
retrieve_credentials_async(req.user_id, req.broker)
for req in requests
]
return await asyncio.gather(*tasks)
# Replace in-memory storage with Vault
class APIKeyManager:
def __init__(self, vault_client=None):
if vault_client is None:
# Initialize Vault client
self.vault = hvac.Client(
url=os.getenv('VAULT_ADDR', 'http://localhost:8200'),
token=os.getenv('VAULT_TOKEN')
)
else:
self.vault = vault_client
logger.info("API key manager initialized with Vault backend")
def store_user_api_key(self, user_id, broker, api_key, api_secret, additional_params=None):
"""Store credentials in Vault instead of memory."""
secret_path = f"secret/users/{user_id}/{broker}"
secret_data = {
'api_key': api_key,
'api_secret': api_secret,
'broker': broker,
'created_at': datetime.now().isoformat()
}
if additional_params:
secret_data['additional_params'] = additional_params
self.vault.secrets.kv.v2.create_or_update_secret(
path=secret_path,
secret=secret_data
)
logger.info(f"Stored credentials in Vault for user {user_id} on {broker}")
def get_user_api_key(self, user_id, broker):
"""Retrieve credentials from Vault."""
secret_path = f"secret/users/{user_id}/{broker}"
try:
response = self.vault.secrets.kv.v2.read_secret_version(
path=secret_path
)
return response['data']['data']
except hvac.exceptions.InvalidPath:
logger.warning(f"No credentials found for user {user_id} on {broker}")
return None
# .env additions
VAULT_ADDR=https://vault.nija.io:8200
VAULT_TOKEN=<vault-token> # For development only
VAULT_ROLE_ID=<approle-id> # For production
VAULT_SECRET_ID=<approle-secret> # For production
VAULT_NAMESPACE=nija # Optional for Vault Enterprise
# docker-compose.yml
services:
vault:
image: vault:1.15
ports:
- "8200:8200"
environment:
VAULT_DEV_ROOT_TOKEN_ID: root # Dev only
cap_add:
- IPC_LOCK
nija-api:
build: .
depends_on:
- vault
environment:
VAULT_ADDR: http://vault:8200
VAULT_TOKEN: root # Dev only
Recommendation: Start with self-hosted for cost efficiency.
Document Version: 1.0 Last Updated: January 27, 2026 Status: ✅ Ready for Implementation Owner: Security Team