Credit Report Agent - Stage 3: Secure¶

class FileValidator:
    """
    ✅ Comprehensive validation pipeline
    """
    def validate(self, file):
        # Layer 1: File extension
        self._validate_extension(file.filename)

        # Layer 2: MIME type
        self._validate_mime_type(file)

        # Layer 3: Magic bytes
        self._validate_magic_bytes(file)

        # Layer 4: File size
        self._validate_size(file)

        # Layer 5: File structure
        self._validate_pdf_structure(file)

        # Layer 6: Content analysis
        self._validate_content(file)

        # Layer 7: Virus scan
        self._scan_for_malware(file)

        # Layer 8: Metadata sanitization
        self._sanitize_metadata(file)

        return True

# Layer 3 example: Magic byte verification
def _validate_magic_bytes(self, file):
    """Verify actual file format"""
    magic = file.read(4)
    file.seek(0)

    if magic != b'%PDF':
        raise ValidationError("Not a valid PDF file")

    # Additional checks for PDF version
    header = file.read(8).decode('latin-1')
    file.seek(0)

    if not re.match(r'%PDF-1\.[0-7]', header):
        raise ValidationError("Unsupported PDF version")

from cryptography.hazmat.primitives.ciphers.aead import AESGCM
import secrets

class FieldEncryption:
    """
    ✅ Encrypt all PII fields individually
    """
    def __init__(self, master_key):
        self.master_key = master_key

    def encrypt_report(self, report):
        """Encrypt every sensitive field"""
        return {
            'ssn': self._encrypt_field(report['ssn']),
            'name': self._encrypt_field(report['name']),
            'address': self._encrypt_field(report['address']),
            'dob': self._encrypt_field(report['dob']),
            'credit_score': self._encrypt_field(str(report['score'])),
            'accounts': [self._encrypt_field(acc) for acc in report['accounts']],
            'inquiries': [self._encrypt_field(inq) for inq in report['inquiries']],
            'employers': [self._encrypt_field(emp) for emp in report['employers']],
            # Metadata not encrypted
            'report_id': report['id'],
            'created_at': report['created_at'],
            'encrypted': True
        }

    def _encrypt_field(self, value):
        """AES-256-GCM encryption with unique nonce"""
        nonce = secrets.token_bytes(12)  # 96-bit nonce
        cipher = AESGCM(self.master_key)
        ciphertext = cipher.encrypt(nonce, value.encode(), None)

        # Return nonce + ciphertext
        return base64.b64encode(nonce + ciphertext).decode()

    def decrypt_field(self, encrypted_value):
        """Decrypt field with authentication"""
        data = base64.b64decode(encrypted_value)
        nonce = data[:12]
        ciphertext = data[12:]

        cipher = AESGCM(self.master_key)
        plaintext = cipher.decrypt(nonce, ciphertext, None)

        return plaintext.decode()

import pyotp
from datetime import datetime, timedelta

class MFAAuthenticator:
    """
    ✅ Two-factor authentication (TOTP)
    """
    def login(self, username, password, mfa_token):
        # Step 1: Verify password
        if not self._verify_password(username, password):
            self._log_failed_attempt(username)
            raise AuthenticationError("Invalid credentials")

        # Step 2: Verify MFA token
        user = self._get_user(username)
        totp = pyotp.TOTP(user['mfa_secret'])

        if not totp.verify(mfa_token, valid_window=1):
            self._log_failed_mfa(username)
            raise AuthenticationError("Invalid MFA token")

        # Step 3: Check account status
        if user['locked_until'] and user['locked_until'] > datetime.now():
            raise AuthenticationError("Account locked")

        # Success - create session
        session = self._create_session(user)
        self._audit_log('successful_login', username)

        return session

    def _verify_password(self, username, password):
        """Constant-time password verification"""
        user = self._get_user(username)

        if not user:
            # Prevent timing attacks
            bcrypt.checkpw(b'dummy', bcrypt.gensalt())
            return False

        return bcrypt.checkpw(
            password.encode(),
            user['password_hash'].encode()
        )

from collections import defaultdict
from threading import Lock
import time

class TokenBucketRateLimiter:
    """
    ✅ Token bucket algorithm for rate limiting
    """
    def __init__(self, rate=10, burst=20):
        self.rate = rate          # Tokens per second
        self.burst = burst        # Max tokens
        self.buckets = defaultdict(lambda: {
            'tokens': burst,
            'last_update': time.time()
        })
        self.lock = Lock()

    def check_limit(self, user_id, cost=1):
        """Check if request allowed"""
        with self.lock:
            bucket = self.buckets[user_id]
            now = time.time()

            # Add tokens based on time elapsed
            elapsed = now - bucket['last_update']
            bucket['tokens'] = min(
                self.burst,
                bucket['tokens'] + elapsed * self.rate
            )
            bucket['last_update'] = now

            # Check if enough tokens
            if bucket['tokens'] >= cost:
                bucket['tokens'] -= cost
                return True

            return False

    def get_retry_after(self, user_id, cost=1):
        """Calculate retry-after time"""
        bucket = self.buckets[user_id]
        tokens_needed = cost - bucket['tokens']

        if tokens_needed <= 0:
            return 0

        return tokens_needed / self.rate

# Usage in endpoint
@app.route('/upload', methods=['POST'])
@require_auth
def upload_report(user_id):
    # Check rate limit
    if not rate_limiter.check_limit(user_id, cost=10):
        retry_after = rate_limiter.get_retry_after(user_id, cost=10)
        return {
            'error': 'Rate limit exceeded',
            'retry_after': retry_after
        }, 429

    return process_upload(request.files['document'])

import json
import hashlib
from datetime import datetime

class AuditLogger:
    """
    ✅ Tamper-proof audit logging
    """
    def __init__(self):
        self.last_hash = None

    def log_event(self, event_type, user_id, resource_id, action, result, metadata=None):
        """
        Log security-relevant event with chaining
        """
        event = {
            'timestamp': datetime.utcnow().isoformat(),
            'event_type': event_type,
            'user_id': user_id,
            'resource_id': resource_id,
            'action': action,
            'result': result,
            'metadata': metadata or {},
            'previous_hash': self.last_hash
        }

        # Calculate hash for integrity
        event_json = json.dumps(event, sort_keys=True)
        event_hash = hashlib.sha256(event_json.encode()).hexdigest()
        event['hash'] = event_hash

        # Store event
        self._store_event(event)

        # Update chain
        self.last_hash = event_hash

        return event_hash

    def verify_chain(self):
        """Verify audit log integrity"""
        events = self._load_all_events()
        prev_hash = None

        for event in events:
            # Check chain
            if event['previous_hash'] != prev_hash:
                return False, f"Chain broken at event {event['timestamp']}"

            # Recalculate hash
            event_copy = event.copy()
            stored_hash = event_copy.pop('hash')
            calculated_hash = hashlib.sha256(
                json.dumps(event_copy, sort_keys=True).encode()
            ).hexdigest()

            if stored_hash != calculated_hash:
                return False, f"Tampered event at {event['timestamp']}"

            prev_hash = stored_hash

        return True, "Audit log intact"

# Usage
audit = AuditLogger()

# Log file access
audit.log_event(
    event_type='file_access',
    user_id='user123',
    resource_id='report456',
    action='view',
    result='success',
    metadata={'ip': request.remote_addr}
)

# Log permission change
audit.log_event(
    event_type='permission_change',
    user_id='admin',
    resource_id='user123',
    action='grant_access',
    result='success',
    metadata={'permission': 'report_admin'}
)

from enum import Enum
from typing import Set

class Permission(Enum):
    """Fine-grained permissions"""
    VIEW_OWN_REPORT = "view_own_report"
    VIEW_ANY_REPORT = "view_any_report"
    UPLOAD_REPORT = "upload_report"
    DELETE_REPORT = "delete_report"
    MANAGE_USERS = "manage_users"
    VIEW_AUDIT_LOG = "view_audit_log"

class Role:
    """Role definitions"""
    USER = {Permission.VIEW_OWN_REPORT, Permission.UPLOAD_REPORT}
    ANALYST = {Permission.VIEW_ANY_REPORT, Permission.VIEW_AUDIT_LOG}
    ADMIN = {Permission.VIEW_ANY_REPORT, Permission.DELETE_REPORT,
             Permission.MANAGE_USERS, Permission.VIEW_AUDIT_LOG}

class AccessControl:
    """
    ✅ Role-based access control
    """
    def check_permission(self, user_id, permission, resource_id=None):
        """Verify user has required permission"""
        user = self._get_user(user_id)
        user_permissions = self._get_permissions(user['role'])

        if permission not in user_permissions:
            self._audit_log('permission_denied', user_id, permission)
            return False

        # Resource-level checks
        if resource_id and permission == Permission.VIEW_OWN_REPORT:
            report = self._get_report(resource_id)
            if report['owner_id'] != user_id:
                self._audit_log('unauthorized_access_attempt', user_id, resource_id)
                return False

        return True

# Usage
@app.route('/report/<report_id>')
@require_auth
def get_report(user_id, report_id):
    if not access_control.check_permission(
        user_id,
        Permission.VIEW_OWN_REPORT,
        resource_id=report_id
    ):
        return {'error': 'Access denied'}, 403

    return fetch_report(report_id)