FinTech Security: Protecting Financial Applications

# Encrypt sensitive data before storage
from cryptography.fernet import Fernet

def encrypt_data(data, key):
    f = Fernet(key)
    return f.encrypt(data.encode())

def decrypt_data(encrypted, key):
    f = Fernet(key)
    return f.decrypt(encrypted).decode()

# MFA requirements
authentication:
  methods:
    - password
    - totp  # Time-based
    - sms    # Less secure
  required_for:
    - transactions
    - settings
    - withdrawals

// Authorization code flow
const authUrl = `https://auth.example.com/authorize?
  client_id=${clientId}
  &redirect_uri=${redirectUri}
  &response_type=code
  &scope=openid profile email
  &state=${state}`;

# Define roles and permissions
ROLES = {
    'admin': ['read', 'write', 'delete', 'admin'],
    'manager': ['read', 'write', 'approve'],
    'user': ['read'],
    'viewer': ['read_own']
}

def check_permission(user_role, action):
    return action in ROLES.get(user_role, [])

// Input validation
function validateAmount(amount) {
  if (typeof amount !== 'number') return false;
  if (amount <= 0) return false;
  if (amount > MAX_TRANSACTION) return false;
  return true;
}

// Parameterized queries (prevent SQL injection)
const query = 'SELECT * FROM accounts WHERE id = ?';
db.execute(query, [accountId]);

# Basic fraud indicators
FRAUD_INDICATORS = [
    'multiple_failures',
    'unusual_location',
    'velocity_exceeded',
    'device_fingerprint_change',
    'amount_anomaly'
]

def calculate_risk_score(transaction):
    score = 0
    if transaction.amount > usual_amount * 5:
        score += 30
    if transaction.failures > 3:
        score += 25
    if location_changed:
        score += 20
    return score

# API rate limits
rate_limits:
  default: 100/minute
  authentication: 5/minute
  transactions: 10/minute
  read: 1000/minute

# Transaction logging
def log_transaction(user_id, amount, action):
    logger.info({
        'timestamp': datetime.utcnow(),
        'user_id': user_id,
        'amount': amount,
        'action': action,
        'ip': request.ip,
        'success': True
    })

# Zero Trust policy configuration
zero_trust:
  identity:
    provider: "Okta/Azure AD"
    mfa_required: true
    device_compliance: "Required"
    session_timeout: 15m

  network_segmentation:
    segmentation: "Micro-perimeter per workload"
    east_west: "Encrypted mTLS required"
    north_south: "API gateway with WAF"

  continuous_verification:
    behavioral_analysis: true
    anomaly_detection: "ML-based"
    reauthentication: "Per sensitive action"

  data_protection:
    encryption: "AES-256 at rest, TLS 1.3 in transit"
    dlp: "Inline data loss prevention"
    tokenization: "For PII and card data"

package middleware

import (
    "crypto/rsa"
    "fmt"
    "net/http"
    "strings"
    "time"

    "github.com/golang-jwt/jwt/v5"
)

type Claims struct {
    Sub       string   `json:"sub"`
    Iss       string   `json:"iss"`
    Aud       []string `json:"aud"`
    Scope     string   `json:"scope"`
    ClientID  string   `json:"client_id"`
    jwt.RegisteredClaims
}

type AuthMiddleware struct {
    publicKey *rsa.PublicKey
    issuer    string
    audience  string
    requiredScope string
}

func NewAuthMiddleware(
    pemBytes []byte, issuer, audience, scope string,
) (*AuthMiddleware, error) {
    key, err := jwt.ParseRSAPublicKeyFromPEM(pemBytes)
    if err != nil {
        return nil, fmt.Errorf("failed to parse public key: %w", err)
    }
    return &AuthMiddleware{
        publicKey:     key,
        issuer:        issuer,
        audience:      audience,
        requiredScope: scope,
    }, nil
}

func (m *AuthMiddleware) ValidateToken(next http.Handler) http.Handler {
    return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
        authHeader := r.Header.Get("Authorization")
        if authHeader == "" {
            http.Error(w, "missing authorization header",
                http.StatusUnauthorized)
            return
        }

        tokenString := strings.TrimPrefix(authHeader, "Bearer ")
        if tokenString == authHeader {
            http.Error(w, "invalid authorization format",
                http.StatusUnauthorized)
            return
        }

        claims := &Claims{}
        token, err := jwt.ParseWithClaims(
            tokenString, claims,
            func(token *jwt.Token) (interface{}, error) {
                if _, ok := token.Method.(*jwt.SigningMethodRSA); !ok {
                    return nil, fmt.Errorf("unexpected signing method: %v",
                        token.Header["alg"])
                }
                return m.publicKey, nil
            },
            jwt.WithIssuer(m.issuer),
            jwt.WithAudience(m.audience),
            jwt.WithLeeway(30*time.Second),
        )
        if err != nil || !token.Valid {
            http.Error(w, "invalid or expired token",
                http.StatusUnauthorized)
            return
        }

        if !strings.Contains(claims.Scope, m.requiredScope) {
            http.Error(w, "insufficient scope",
                http.StatusForbidden)
            return
        }

        ctx := context.WithValue(r.Context(), "claims", claims)
        next.ServeHTTP(w, r.WithContext(ctx))
    })
}

# Cloud security monitoring configuration
cloud_security:
  providers:
    - aws
    - azure
    - gcp

  cspm_tool: "Wiz / Prisma Cloud / CrowdStrike Falcon"

  monitoring_rules:
    - name: "S3 Public Access"
      severity: "Critical"
      auto_remediate: true
      action: "Block public ACLs"

    - name: "IAM Overprivileged Roles"
      severity: "High"
      auto_remediate: false
      action: "Alert security team"

    - name: "Unencrypted Databases"
      severity: "Critical"
      auto_remediate: true
      action: "Enable encryption at rest"

    - name: "Network Exposure"
      severity: "High"
      auto_remediate: false
      action: "Alert and restrict ingress rules"

  compliance_frameworks:
    - "PCI DSS 4.0"
    - "SOC 2 Type II"
    - "ISO 27001"
    - "GDPR"

#!/usr/bin/env python3
"""Automated secrets rotation for machine identities."""

import os
import json
import base64
import logging
from datetime import datetime, timedelta
from typing import Dict, List

import boto3
import requests
from cryptography.fernet import Fernet
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.asymmetric import rsa

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


class SecretsRotationManager:
    """Manage automated rotation of machine secrets and certificates."""

    def __init__(self, vault_url: str, vault_token: str):
        self.vault_url = vault_url
        self.vault_token = vault_token
        self.session = requests.Session()
        self.session.headers.update({
            "X-Vault-Token": vault_token
        })

    def rotate_database_password(
        self, db_instance: str, rotation_window: int = 24
    ) -> bool:
        """Rotate database credentials with zero downtime."""
        try:
            # Generate new password
            new_password = base64.b64encode(os.urandom(24)).decode()

            # Update database
            response = self.session.post(
                f"{self.vault_url}/v1/database/rotate-root/{db_instance}"
            )
            response.raise_for_status()

            # Verify new credentials
            if self._verify_db_connection(db_instance):
                logger.info(
                    f"Rotated password for {db_instance}"
                )
                return True
            else:
                logger.error(
                    f"Rotation failed verification for {db_instance}"
                )
                return False

        except Exception as e:
            logger.error(f"Rotation error for {db_instance}: {e}")
            return False

    def rotate_api_key(self, service_name: str, key_id: str) -> str:
        """Generate a new API key and invalidate the old one."""
        new_key = base64.b64encode(os.urandom(32)).decode()

        payload = {
            "key_id": key_id,
            "new_key": new_key,
            "rotated_at": datetime.utcnow().isoformat(),
            "issued_by": "rotation-manager",
        }

        response = self.session.post(
            f"{self.vault_url}/v1/secret/data/{service_name}",
            json={"data": payload},
        )
        response.raise_for_status()
        logger.info(f"Rotated API key for {service_name}")
        return new_key

    def rotate_tls_certificate(
        self, domain: str, validity_days: int = 365
    ) -> Dict:
        """Generate a new TLS certificate for internal services."""
        private_key = rsa.generate_private_key(
            public_exponent=65537,
            key_size=4096,
        )

        rotation_record = {
            "domain": domain,
            "rotated_at": datetime.utcnow().isoformat(),
            "expires_at": (
                datetime.utcnow() + timedelta(days=validity_days)
            ).isoformat(),
            "key_algorithm": "RSA-4096",
        }

        response = self.session.post(
            f"{self.vault_url}/v1/pki/issue/{domain}",
            json={
                "common_name": domain,
                "ttl": f"{validity_days}h",
            },
        )
        response.raise_for_status()
        logger.info(
            f"Rotated TLS certificate for {domain}"
        )
        return rotation_record

    def _verify_db_connection(self, db_instance: str) -> bool:
        """Verify database connectivity with rotated credentials."""
        try:
            # Implement actual database connection verification
            return True
        except Exception:
            return False

# .gitlab-ci.yml - security-integrated pipeline
stages:
  - lint
  - security-scan
  - test
  - build
  - deploy

variables:
  DOCKER_IMAGE: $CI_REGISTRY_IMAGE:$CI_COMMIT_SHORT_SHA

security-lint:
  stage: lint
  script:
    - semgrep --config=auto --error .
    - bandit -r src/ -ll
  artifacts:
    reports:
      sast: gl-sast-report.json

secret-scanning:
  stage: security-scan
  script:
    - trufflehog filesystem --directory=. --json
    - gitleaks detect --verbose
  artifacts:
    paths:
      - gitleaks-report.json

dependency-scan:
  stage: security-scan
  script:
    - trivy filesystem --severity CRITICAL,HIGH --exit-code 1 .
    - pip-audit --desc on
  artifacts:
    reports:
      dependency_scanning: gl-dependency-scanning-report.json

container-scan:
  stage: security-scan
  script:
    - docker build -t $DOCKER_IMAGE .
    - trivy image --severity CRITICAL,HIGH --exit-code 1 $DOCKER_IMAGE
  artifacts:
    reports:
      container_scanning: gl-container-scanning-report.json

dynamic-analysis:
  stage: security-scan
  script:
    - docker-compose up -d
    - zap-cli --config zap.conf quick-scan http://localhost:8080
  after_script:
    - docker-compose down

deploy-production:
  stage: deploy
  script:
    - kubectl apply -f k8s/manifests/
  environment: production
  rules:
    - if: $CI_COMMIT_BRANCH == "main"

incident_response:
  playbook_template:
    id: "IR-001"
    name: "Data Breach - Customer PII Exposure"

    detection:
      sources:
        - DLP alert
        - Customer report
        - SOC escalation
      validation_steps:
        - Confirm alert with SIEM logs
        - Check affected data classification
        - Identify data access pattern

    containment:
      immediate:
        - Revoke affected IAM credentials
        - Block source IPs in WAF
        - Enable read-only mode for affected services
      short_term:
        - Rotate database credentials
        - Enable query logging
        - Apply network isolation

    communication:
      internal:
        - Alert CISO and Legal
        - Incident channel on Slack
        - 15-minute status updates
      external:
        - Legal counsel notification
        - Cyber insurance carrier
        - Regulatory bodies (timeline per jurisdiction)
        - Affected customers (template approved by Legal)

    forensics:
      data_collection:
        - CloudTrail / Azure Activity Logs
        - Database audit logs
        - Application server logs
        - Network flow logs
      analysis:
        - Timeline reconstruction
        - Data exfiltration assessment
        - Root cause identification

    recovery:
      steps:
        - Patch identified vulnerabilities
        - Restore from pre-incident backup
        - Verify security control effectiveness
        - Gradual service restoration with monitoring

    post_mortem:
      deliverables:
        - Root cause analysis document
        - Remediation roadmap
        - Updated security controls
        - Executive summary for board