Task Collaboration Agent - Stage 3: Secure¶

import secrets

class SessionManager:
    """
    ✅ Production session management
    """
    def create_session(self, user_id, client_context):
        # 256-bit cryptographically random ID
        session_id = secrets.token_urlsafe(32)

        session_data = {
            'user_id': user_id,
            'created_at': time.time(),
            'last_activity': time.time(),
            'ip_address': client_context['ip'],
            'user_agent': client_context['user_agent'],
            'client_cert_hash': client_context['cert_hash'],
            'mfa_verified': True,
            'permissions': self._get_user_permissions(user_id)
        }

        # Encrypt session state
        encrypted_state = self.encryptor.encrypt(session_data)

        # Store encrypted
        self.sessions[session_id] = encrypted_state

        self.audit.log('session_created', user_id, session_id)

        return session_id

import pyotp
import bcrypt

class AuthManager:
    """
    ✅ Two-factor authentication
    """
    def authenticate(self, username, password, mfa_token):
        # Rate limit check
        if not self.rate_limiter.check(username):
            self.audit.log('rate_limit_exceeded', username)
            raise AuthenticationError("Too many attempts")

        # Step 1: Verify password
        if not self._verify_password(username, password):
            self.audit.log('auth_failed_password', username)
            self._increment_failed_attempts(username)
            raise AuthenticationError("Invalid credentials")

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

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

        # Success - reset failed attempts
        self._reset_failed_attempts(username)
        self.audit.log('auth_success', username)

        return user['id']

    def _verify_password(self, username, password):
        """Constant-time password verification"""
        if username not in self.users:
            # Prevent timing attacks
            bcrypt.checkpw(b'dummy', bcrypt.gensalt())
            return False

        user = self.users[username]
        return bcrypt.checkpw(
            password.encode(),
            user['password_hash'].encode()
        )

class SessionValidator:
    """
    ✅ Comprehensive session validation
    """
    def validate_session(self, session_id, request_context):
        # Decrypt session state
        encrypted = self.sessions.get(session_id)
        if not encrypted:
            return False

        try:
            session = self.encryptor.decrypt(encrypted)
        except DecryptionError:
            self.audit.log('session_decrypt_failed', session_id)
            return False

        # Check idle timeout (30 minutes)
        idle_time = time.time() - session['last_activity']
        if idle_time > self.IDLE_TIMEOUT:
            self._invalidate_session(session_id, 'idle_timeout')
            return False

        # Check absolute timeout (8 hours)
        session_age = time.time() - session['created_at']
        if session_age > self.ABSOLUTE_TIMEOUT:
            self._invalidate_session(session_id, 'absolute_timeout')
            return False

        # Enforce binding factors
        if not self._validate_bindings(session, request_context):
            self._invalidate_session(session_id, 'binding_violation')
            self.security_monitor.alert('session_binding_violation', {
                'session_id': session_id,
                'expected': session,
                'actual': request_context
            })
            return False

        # Check MFA verified
        if not session.get('mfa_verified'):
            return False

        # Update last activity
        session['last_activity'] = time.time()
        self.sessions[session_id] = self.encryptor.encrypt(session)

        return True

    def _validate_bindings(self, session, context):
        """Validate all binding factors"""
        # IP address binding
        if session['ip_address'] != context['ip']:
            return False

        # User agent binding
        if session['user_agent'] != context['user_agent']:
            return False

        # Client certificate binding
        if session['client_cert_hash'] != context['cert_hash']:
            return False

        return True

import hashlib
from collections import deque
from threading import Lock

class NonceValidator:
    """
    ✅ Prevent replay attacks
    """
    def __init__(self, window_size=300):  # 5 minutes
        self.window_size = window_size
        self.seen_nonces = deque(maxlen=10000)
        self.nonce_timestamps = {}
        self.lock = Lock()

    def validate(self, nonce, timestamp):
        """
        Validate nonce is unique and recent
        """
        with self.lock:
            current_time = time.time()

            # Check timestamp is recent
            if abs(current_time - timestamp) > self.window_size:
                return False

            # Check nonce hasn't been seen
            nonce_key = hashlib.sha256(
                f"{nonce}:{timestamp}".encode()
            ).hexdigest()

            if nonce_key in self.seen_nonces:
                # Replay attack detected!
                self.audit.log('replay_attack_detected', {
                    'nonce': nonce,
                    'timestamp': timestamp
                })
                return False

            # Record nonce
            self.seen_nonces.append(nonce_key)
            self.nonce_timestamps[nonce_key] = timestamp

            # Cleanup old nonces
            self._cleanup_old_nonces(current_time)

            return True

    def _cleanup_old_nonces(self, current_time):
        """Remove nonces outside time window"""
        cutoff = current_time - self.window_size

        to_remove = [
            nonce for nonce, ts in self.nonce_timestamps.items()
            if ts < cutoff
        ]

        for nonce in to_remove:
            del self.nonce_timestamps[nonce]

# Usage in message handler
def handle_message(message, signature):
    nonce = message.get('nonce')
    timestamp = message.get('timestamp')

    # Verify nonce (prevent replay)
    if not nonce_validator.validate(nonce, timestamp):
        return error("Invalid or replayed nonce")

    # Verify HMAC signature
    if not verify_hmac(message, signature):
        return error("Invalid signature")

    # Process message (safe from replay)
    return process_message(message)

from collections import defaultdict
from threading import Lock

class TokenBucketRateLimiter:
    """
    ✅ Production 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, identifier, cost=1):
        """Check if request allowed"""
        with self.lock:
            bucket = self.buckets[identifier]
            now = time.time()

            # Add tokens based on elapsed time
            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

            # Log rate limit exceeded
            self.audit.log('rate_limit_exceeded', identifier)
            return False

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

        if tokens_needed <= 0:
            return 0

        return tokens_needed / self.rate

# Usage in endpoints
@require_auth
def handle_request(user_id, request):
    # Check rate limit
    if not rate_limiter.check_limit(user_id, cost=5):
        retry_after = rate_limiter.get_retry_after(user_id, cost=5)
        return {
            'error': 'Rate limit exceeded',
            'retry_after': retry_after
        }, 429

    return process_request(request)

import ssl

class TLSConfig:
    """
    ✅ Production TLS configuration
    """
    @staticmethod
    def create_server_context(certfile, keyfile, cafile):
        """Create TLS 1.3 server context"""
        context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)

        # Require TLS 1.3
        context.minimum_version = ssl.TLSVersion.TLSv1_3
        context.maximum_version = ssl.TLSVersion.TLSv1_3

        # Load server certificate
        context.load_cert_chain(certfile, keyfile)

        # Enable mutual TLS (client certs required)
        context.verify_mode = ssl.CERT_REQUIRED
        context.load_verify_locations(cafile)

        # Strong ciphers only
        context.set_ciphers('TLS_AES_256_GCM_SHA384:TLS_CHACHA20_POLY1305_SHA256')

        return context

    @staticmethod
    def create_client_context(certfile, keyfile, cafile):
        """Create TLS 1.3 client context"""
        context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)

        # Require TLS 1.3
        context.minimum_version = ssl.TLSVersion.TLSv1_3

        # Load client certificate
        context.load_cert_chain(certfile, keyfile)

        # Verify server certificate
        context.verify_mode = ssl.CERT_REQUIRED
        context.load_verify_locations(cafile)

        # Strong ciphers only
        context.set_ciphers('TLS_AES_256_GCM_SHA384:TLS_CHACHA20_POLY1305_SHA256')

        return context

# Usage
ssl_context = TLSConfig.create_server_context(
    certfile='certs/coordinator.crt',
    keyfile='certs/coordinator.key',
    cafile='certs/ca.crt'
)

# Wrap socket with TLS
secure_socket = ssl_context.wrap_socket(
    raw_socket,
    server_side=True
)

from enum import Enum
from typing import Set

class Permission(Enum):
    """Fine-grained permissions"""
    CREATE_PROJECT = "create_project"
    DELETE_PROJECT = "delete_project"
    VIEW_PROJECT = "view_project"
    MODIFY_PROJECT = "modify_project"
    CREATE_TASK = "create_task"
    ASSIGN_TASK = "assign_task"
    COMPLETE_TASK = "complete_task"
    VIEW_AUDIT_LOG = "view_audit_log"
    MANAGE_USERS = "manage_users"

class Role:
    """Role definitions"""
    VIEWER = {Permission.VIEW_PROJECT}
    CONTRIBUTOR = {
        Permission.VIEW_PROJECT,
        Permission.CREATE_TASK,
        Permission.COMPLETE_TASK
    }
    PROJECT_ADMIN = {
        Permission.CREATE_PROJECT,
        Permission.DELETE_PROJECT,
        Permission.VIEW_PROJECT,
        Permission.MODIFY_PROJECT,
        Permission.CREATE_TASK,
        Permission.ASSIGN_TASK,
        Permission.COMPLETE_TASK
    }
    SYSTEM_ADMIN = set(Permission)  # All permissions

class RBACManager:
    """
    ✅ Role-based access control
    """
    def check_permission(self, user_id, permission, resource_id=None):
        """Verify user has required permission"""
        user = self.users[user_id]
        user_permissions = self._get_effective_permissions(user)

        # Check permission exists
        if permission not in user_permissions:
            self.audit.log('permission_denied', {
                'user_id': user_id,
                'permission': permission.value,
                'resource_id': resource_id
            })
            return False

        # Resource-level checks
        if resource_id:
            if not self._check_resource_access(user_id, resource_id):
                self.audit.log('resource_access_denied', {
                    'user_id': user_id,
                    'resource_id': resource_id
                })
                return False

        return True

    def _get_effective_permissions(self, user) -> Set[Permission]:
        """Get all user permissions (role + explicit)"""
        permissions = set()

        # Add role permissions
        for role in user['roles']:
            permissions.update(Role[role].value)

        # Add explicit permissions
        permissions.update(user.get('explicit_permissions', []))

        return permissions

# Usage
@require_auth
def create_project(session_id, project_data):
    session = get_session(session_id)

    # Check permission
    if not rbac.check_permission(
        session['user_id'],
        Permission.CREATE_PROJECT
    ):
        return {'error': 'Access denied'}, 403

    # Create project
    project = create_project_internal(project_data)
    return {'status': 'success', 'project': project}

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

class StateEncryptor:
    """
    ✅ AES-256-GCM state encryption
    """
    def __init__(self, master_key):
        self.cipher = AESGCM(master_key)  # 256-bit key

    def encrypt(self, data):
        """Encrypt session/state data"""
        # Serialize data
        plaintext = json.dumps(data).encode()

        # Generate unique nonce (96 bits)
        nonce = secrets.token_bytes(12)

        # Encrypt with authentication
        ciphertext = self.cipher.encrypt(nonce, plaintext, None)

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

    def decrypt(self, encrypted_data):
        """Decrypt and verify state data"""
        try:
            # Decode
            data = base64.b64decode(encrypted_data)

            # Split nonce and ciphertext
            nonce = data[:12]
            ciphertext = data[12:]

            # Decrypt and verify
            plaintext = self.cipher.decrypt(nonce, ciphertext, None)

            # Deserialize
            return json.loads(plaintext.decode())

        except Exception as e:
            # Decryption failure = tampering
            raise DecryptionError("State tampered or corrupted")

import hashlib
import json

class TamperProofAuditLogger:
    """
    ✅ Tamper-proof audit logging with chaining
    """
    def __init__(self):
        self.last_hash = None
        self.log_file = 'audit.log'

    def log(self, event_type, user_id, details):
        """Log security event with chain integrity"""
        event = {
            'timestamp': time.time(),
            'event_type': event_type,
            'user_id': user_id,
            'details': details,
            'previous_hash': self.last_hash
        }

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

        # Store event
        with open(self.log_file, 'a') as f:
            f.write(json.dumps(event) + '\n')

        # Update chain
        self.last_hash = event_hash

        return event_hash

    def verify_integrity(self):
        """Verify audit log hasn't been tampered"""
        with open(self.log_file, 'r') as f:
            events = [json.loads(line) for line in f]

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

            # Verify hash
            stored_hash = event.pop('hash')
            calculated = hashlib.sha256(
                json.dumps(event, sort_keys=True).encode()
            ).hexdigest()

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

            prev_hash = stored_hash

        return True, "Audit log intact"