FinTech Regulatory Security — SEBI CSCRF, CERT-In Auditing, Secrets Rotation

Security compliance for regulated financial entities in India — SEBI-registered brokers, exchanges, depository participants, and Asset Management Companies (AMCs).

The Regulatory Landscape

SEBI (Securities and Exchange Board of India)
  └── Issues: CSCRF (Cybersecurity and Cyber Resilience Framework)
        └── Mandates: risk management, incident response, audits

CERT-In (Indian Computer Emergency Response Team, under MeitY)
  └── Empanels: security audit firms
  └── Mandates: 6-hour incident reporting, log retention, VAPT audits

RBI (Reserve Bank of India)
  └── Issues: guidelines for payment systems, NBFC cybersecurity

SEBI CSCRF (2023, updated 2024) is the primary framework for capital market entities. Non-compliance can result in suspension of trading licenses.

1. SEBI CSCRF — Key Controls

Risk Categories and Classification

SEBI classifies entities into three categories based on trading volume and systemic importance:

Mandatory Controls Under CSCRF

1. Network Segmentation

Production trading systems must be isolated:

  Internet Zone
      │
      ▼ DMZ (WAF, reverse proxy)
      │
      ▼ Application Zone (OMS, RMS)
      │
      ▼ Database Zone (trade DB, position DB)
      │
      ▼ Exchange Connectivity Zone (FIX gateway, co-lo)

No direct path from Internet Zone to Database Zone.
Each crossing requires firewall with explicit allow rules.
Rules documented and reviewed quarterly.

2. Privileged Access Management (PAM)

All admin access to production must go through a PAM tool (CyberArk, BeyondTrust, or open-source Teleport):

# Teleport (open-source PAM) setup example
# All SSH and kubectl access is recorded and auditable

# Instead of: ssh admin@prod-db-01
# Engineers use:
tsh ssh admin@prod-db-01
# → session recorded, stored in audit log with timestamp
# → every command typed is captured

# kubectl access
tsh kube login prod-cluster
kubectl get pods
# → all API calls logged with user identity + timestamp

3. Multi-Factor Authentication

All access to trading systems, admin consoles, and cloud portals must require MFA. Hardcoded in IAM policies:

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Deny",
      "Action": "*",
      "Resource": "*",
      "Condition": {
        "BoolIfExists": {
          "aws:MultiFactorAuthPresent": "false"
        }
      }
    }
  ]
}

4. Data Localisation

SEBI requires trade data, investor data, and audit logs to remain on servers physically located in India. Cloud setup must use ap-south-1 (Mumbai) or on-prem in India. Replication to overseas regions requires approval and data masking.

2. CERT-In Compliance — Evidence-Based Auditing

What CERT-In Auditors Look For

CERT-In empaneled auditors don't just look at your config files — they require documented, timestamped evidence that controls are actually operational. Kubernetes RBAC policies on a slide deck are not evidence.

What auditors actually want:
  ✓  Screenshot + timestamp of who has admin access (from PAM tool)
  ✓  Log showing the last 90 days of admin logins and commands run
  ✓  Evidence that RBAC was reviewed on [specific date] by [specific person]
  ✓  Change tickets linked to every firewall rule modification
  ✓  Vulnerability scan report dated within 6 months (VAPT)
  ✓  Incident response drill conducted on [date] with outcomes documented
  ✓  Proof that terminated employees' access was revoked within 24 hours

Kubernetes RBAC Is Not Enough

A ClusterRole with verbs: [get, list] is a good start, but it doesn't satisfy audit requirements alone. You need:

a) Immutable audit logs of every API call:

# kube-apiserver audit policy — log everything relevant
apiVersion: audit.k8s.io/v1
kind: Policy
rules:
  # Log all secret/configmap access (data access by humans)
  - level: RequestResponse
    users: ["*"]
    resources:
      - group: ""
        resources: ["secrets", "configmaps"]
  
  # Log all admin actions with full request body
  - level: RequestResponse
    userGroups: ["system:masters", "cluster-admin"]
    omitStages: ["RequestReceived"]
  
  # Log all RBAC changes
  - level: RequestResponse
    resources:
      - group: "rbac.authorization.k8s.io"
        resources: ["roles", "clusterroles", "rolebindings", "clusterrolebindings"]
  
  # Log pod exec (someone SSHing into a container)
  - level: RequestResponse
    resources:
      - group: ""
        resources: ["pods/exec", "pods/attach", "pods/portforward"]
  
  # Default: metadata only (who did what, not full body)
  - level: Metadata

# Store audit logs to CloudWatch or S3 with tamper-proof retention
# kube-apiserver flags:
--audit-log-path=/var/log/kubernetes/audit.log
--audit-log-maxage=90          # retain 90 days locally
--audit-log-maxbackup=10
--audit-log-maxsize=100        # 100MB per file
--audit-policy-file=/etc/kubernetes/audit-policy.yaml

# Ship to S3 (immutable, versioned)
# Use Fluent Bit to forward /var/log/kubernetes/audit.log → S3 bucket
# S3 bucket: enable versioning + Object Lock (WORM) for audit integrity

b) Access review reports:

# Generate a timestamped report of who has what access
# Run monthly and store output as evidence

DATE=$(date +%Y-%m-%d)
REPORT_FILE="access-review-${DATE}.txt"

echo "=== Kubernetes RBAC Access Review: ${DATE} ===" > $REPORT_FILE
echo "" >> $REPORT_FILE

echo "--- ClusterRoleBindings (cluster-wide admin access) ---" >> $REPORT_FILE
kubectl get clusterrolebindings -o json | \
  jq -r '.items[] | select(.roleRef.name == "cluster-admin") | 
    "ClusterRoleBinding: \(.metadata.name)\n  Subjects: \(.subjects)"' >> $REPORT_FILE

echo "" >> $REPORT_FILE
echo "--- All RoleBindings in production namespace ---" >> $REPORT_FILE
kubectl get rolebindings -n production -o wide >> $REPORT_FILE

echo "" >> $REPORT_FILE
echo "--- Service accounts with auto-mounted tokens ---" >> $REPORT_FILE
kubectl get serviceaccounts -A -o json | \
  jq -r '.items[] | select(.automountServiceAccountToken != false) | 
    "\(.metadata.namespace)/\(.metadata.name)"' >> $REPORT_FILE

# Upload to evidence bucket
aws s3 cp $REPORT_FILE s3://audit-evidence-bucket/rbac/${DATE}/
echo "Uploaded: s3://audit-evidence-bucket/rbac/${DATE}/${REPORT_FILE}"

c) Change management evidence:

Every firewall rule change, RBAC modification, or security group update must have a change ticket (Jira, ServiceNow) referenced in the commit or API call. Achieve this via:

# Require change ticket in Terraform commit message (pre-commit hook)
cat .git/hooks/commit-msg
#!/bin/bash
COMMIT_MSG=$(cat "$1")
if ! echo "$COMMIT_MSG" | grep -qE "CHG-[0-9]+|HOTFIX-[0-9]+"; then
  echo "ERROR: Commit message must reference a change ticket (e.g., CHG-1234)"
  exit 1
fi

6-Hour Incident Reporting to CERT-In

CERT-In's 2022 directive requires reporting cybersecurity incidents within 6 hours of detection. This means your alerting and escalation pipeline must be fast:

Alert fires (CloudWatch / Prometheus)
    │ < 5 minutes
    ▼
On-call engineer notified (PagerDuty)
    │ < 30 minutes
    ▼
Incident declared, IR team assembled
    │ < 2 hours
    ▼
Initial assessment: is it a reportable incident?
    │ < 4 hours
    ▼
Report filed with CERT-In portal (https://www.cert-in.org.in)
    │ within 6 hours of detection
    ▼
Internal post-incident report: within 14 days

# Automate CERT-In reportable incident detection via CloudTrail + Lambda
# Trigger on: unauthorized API calls, root login, IAM policy changes, S3 public access

# CloudTrail → EventBridge rule → Lambda → creates Jira ticket + notifies IR team
aws events put-rule \
  --name cert-in-incident-trigger \
  --event-pattern '{
    "source": ["aws.cloudtrail"],
    "detail-type": ["AWS API Call via CloudTrail"],
    "detail": {
      "errorCode": ["AccessDenied", "UnauthorizedAccess"],
      "userIdentity": {
        "type": ["Root"]
      }
    }
  }'

3. Zero-Downtime Secrets Rotation

The Problem

A trading system is stateful. The order management system has live TCP connections to the database. If you rotate the DB password:

Naive rotation:
  1. Change DB password in RDS
  2. Update secret in Secrets Manager
  3. Restart pods to pick up new secret
  → 10–30 second outage
  → Live orders in-flight are lost
  → Exchange may penalize for erratic order flow

For a system handling thousands of orders per second, a 10-second outage means open positions exposed to market risk. Zero-downtime rotation requires the DB to accept both old and new passwords simultaneously during the transition window.

Zero-Downtime Rotation: AWS Secrets Manager

AWS Secrets Manager supports a rotation strategy where the DB accepts both credentials during rotation:

Phase 1: Create new secret version (AWSPENDING)
Phase 2: Update DB to accept BOTH old and new passwords
Phase 3: Verify new password works
Phase 4: Mark new version as AWSCURRENT
Phase 5: Delete old password from DB

# Lambda rotation function (Python)
import boto3
import json

def lambda_handler(event, context):
    arn = event['SecretId']
    token = event['ClientRequestToken']
    step = event['Step']

    service_client = boto3.client('secretsmanager')
    rds_client = boto3.client('rds')

    if step == "createSecret":
        # Generate new credentials, store as AWSPENDING
        current_secret = json.loads(
            service_client.get_secret_value(SecretId=arn, VersionStage="AWSCURRENT")['SecretString']
        )
        new_password = generate_strong_password()
        new_secret = {**current_secret, "password": new_password}
        
        service_client.put_secret_value(
            SecretId=arn,
            ClientRequestToken=token,
            SecretString=json.dumps(new_secret),
            VersionStages=['AWSPENDING']
        )

    elif step == "setSecret":
        # Add NEW password to DB while OLD password still works
        # For PostgreSQL: ALTER USER trading_app WITH PASSWORD 'new_password';
        # For the brief overlap, DB accepts BOTH via a temp dual-auth approach
        current = get_secret(arn, "AWSCURRENT")
        pending = get_secret(arn, "AWSPENDING")
        
        conn = connect_db(current)  # connect with old password
        conn.execute(f"ALTER USER {pending['username']} WITH PASSWORD '{pending['password']}'")
        # ← DB now accepts new password. Old connections still work (existing sessions)

    elif step == "testSecret":
        # Verify new credentials actually work
        pending = get_secret(arn, "AWSPENDING")
        test_conn = connect_db(pending)  # must succeed
        test_conn.execute("SELECT 1")

    elif step == "finishSecret":
        # Promote AWSPENDING → AWSCURRENT
        service_client.update_secret_version_stage(
            SecretId=arn,
            VersionStage="AWSCURRENT",
            MoveToVersionId=token,
            RemoveFromVersionId=get_current_version_id(arn)
        )
        # Existing connections using old password still work until they reconnect
        # New connections use the new password from Secrets Manager

Zero-Downtime Rotation: HashiCorp Vault Dynamic Secrets

Vault's dynamic secrets engine is better than static rotation — it generates unique credentials per service instance, with a TTL. No "rotation event" exists — credentials expire naturally.

sequenceDiagram
    participant APP as Trading App Pod
    participant AGENT as Vault Agent Sidecar
    participant VAULT as HashiCorp Vault
    participant DB as PostgreSQL

    APP->>AGENT: start up
    AGENT->>VAULT: authenticate (Kubernetes auth, using K8s SA token)
    VAULT->>VAULT: verify K8s SA is allowed to get db/creds/trading-role
    VAULT->>DB: CREATE USER vault_abc123 WITH PASSWORD 'x9k2...' VALID UNTIL '...'
    VAULT-->>AGENT: {username: vault_abc123, password: x9k2..., ttl: 1h}
    AGENT-->>APP: write credentials to file /vault/secrets/db-creds

    Note over APP,DB: App connects using vault_abc123

    Note over AGENT,VAULT: At 50% of TTL (30 min), Vault renews

    AGENT->>VAULT: renew lease for vault_abc123
    VAULT-->>AGENT: ttl extended to 1h from now

    Note over AGENT,VAULT: If app is shut down, lease is revoked

    AGENT->>VAULT: revoke lease
    VAULT->>DB: DROP USER vault_abc123

# Setup Vault dynamic secrets for PostgreSQL
vault secrets enable database

vault write database/config/trading-db \
  plugin_name=postgresql-database-plugin \
  allowed_roles="trading-app" \
  connection_url="postgresql://{{username}}:{{password}}@prod-db:5432/trades" \
  username="vault-admin" \
  password="admin-password"

# Define role: what SQL to run to create/revoke credentials
vault write database/roles/trading-app \
  db_name=trading-db \
  creation_statements="
    CREATE ROLE \"{{name}}\" WITH LOGIN PASSWORD '{{password}}' VALID UNTIL '{{expiration}}';
    GRANT SELECT, INSERT, UPDATE ON ALL TABLES IN SCHEMA public TO \"{{name}}\";
  " \
  revocation_statements="DROP ROLE IF EXISTS \"{{name}}\";" \
  default_ttl="1h" \
  max_ttl="24h"

# Kubernetes: Vault Agent Sidecar injection
apiVersion: apps/v1
kind: Deployment
metadata:
  name: order-management
spec:
  template:
    metadata:
      annotations:
        vault.hashicorp.com/agent-inject: "true"
        vault.hashicorp.com/role: "trading-app"
        vault.hashicorp.com/agent-inject-secret-db-creds: "database/creds/trading-app"
        vault.hashicorp.com/agent-inject-template-db-creds: |
          {{- with secret "database/creds/trading-app" -}}
          DB_USERNAME={{ .Data.username }}
          DB_PASSWORD={{ .Data.password }}
          {{- end }}
    spec:
      serviceAccountName: trading-app-sa  # bound to Vault role
      containers:
      - name: order-management
        image: trading/oms:latest
        # App reads /vault/secrets/db-creds, Vault agent updates it before TTL expiry
        # No restart needed — app re-reads file on next connection attempt

Application-Side: Connection Pool Rotation Without Restart

The app must handle credential rotation without dropping live connections:

# Python: PgBouncer + connection pool with credential refresh
import psycopg2
import threading
import time
import os

class RefreshingConnectionPool:
    def __init__(self, creds_file="/vault/secrets/db-creds"):
        self.creds_file = creds_file
        self.pool = None
        self.lock = threading.Lock()
        self._refresh_credentials()
        # Background thread refreshes every 5 minutes
        threading.Thread(target=self._background_refresh, daemon=True).start()

    def _read_creds(self):
        creds = {}
        with open(self.creds_file) as f:
            for line in f:
                k, v = line.strip().split("=", 1)
                creds[k] = v
        return creds

    def _refresh_credentials(self):
        creds = self._read_creds()
        new_pool = create_pool(
            host=os.environ["DB_HOST"],
            user=creds["DB_USERNAME"],
            password=creds["DB_PASSWORD"],
            database="trades"
        )
        with self.lock:
            old_pool = self.pool
            self.pool = new_pool
            if old_pool:
                # Drain old pool — existing queries finish, then connections close
                old_pool.closeall()  # closes after current transactions complete

    def _background_refresh(self):
        while True:
            time.sleep(300)   # check every 5 minutes
            self._refresh_credentials()

    def get_connection(self):
        with self.lock:
            return self.pool.getconn()

Audit Evidence Checklist (CERT-In / SEBI)

SEBI CSCRF Incident Classification