Security Best Practices

# TLS configuration for Cloud Load Balancer
resource "google_compute_ssl_certificate" "neurascale" {
  name_prefix = "neurascale-ssl"
  private_key = file("private-key.pem")
  certificate = file("certificate.pem")
 
  lifecycle {
    create_before_destroy = true
  }
}
 
resource "google_compute_target_https_proxy" "neurascale" {
  name             = "neurascale-https-proxy"
  url_map          = google_compute_url_map.neurascale.id
  ssl_certificates = [google_compute_ssl_certificate.neurascale.id]
 
  # TLS 1.3 only
  ssl_policy = google_compute_ssl_policy.modern.id
}
 
resource "google_compute_ssl_policy" "modern" {
  name            = "neurascale-modern-tls"
  profile         = "MODERN"
  min_tls_version = "TLS_1_2"  # Minimum TLS 1.2
}

# Resource-based authorization
class ResourcePolicy:
    def __init__(self, user: User):
        self.user = user
 
    def can_access_session(self, session: Session) -> bool:
        """Check if user can access specific session"""
 
        # Admin can access all sessions
        if self.user.role == Role.ADMIN:
            return True
 
        # Clinician can access sessions they created or are assigned to
        if self.user.role == Role.CLINICIAN:
            return (session.created_by == self.user.id or
                   self.user.id in session.assigned_clinicians)
 
        # Researcher can access sessions in their studies
        if self.user.role == Role.RESEARCHER:
            return session.study_id in self.user.authorized_studies
 
        # Patient can only access their own sessions
        if self.user.role == Role.PATIENT:
            return session.patient_id == self.user.patient_profile.id
 
        return False
 
    def can_export_data(self, session: Session) -> bool:
        """Check if user can export session data"""
 
        # Must have basic session access first
        if not self.can_access_session(session):
            return False
 
        # Additional restrictions for data export
        if self.user.role == Role.PATIENT:
            return False  # Patients cannot export raw data
 
        # Check data use agreement
        if not self.user.has_signed_data_use_agreement():
            return False
 
        return True
 
    def get_data_access_level(self, session: Session) -> str:
        """Determine level of data access for session"""
 
        if not self.can_access_session(session):
            return "none"
 
        if self.user.role in [Role.ADMIN, Role.CLINICIAN]:
            return "full"  # Access to raw data and PII
 
        if self.user.role == Role.RESEARCHER:
            return "de_identified"  # De-identified data only
 
        if self.user.role == Role.PATIENT:
            return "summary"  # High-level summaries only
 
        return "none"

# Secure VPC setup
resource "google_compute_network" "neurascale_vpc" {
  name                    = "neurascale-vpc"
  auto_create_subnetworks = false
}
 
# Private subnet for application servers
resource "google_compute_subnetwork" "private_subnet" {
  name          = "neurascale-private"
  ip_cidr_range = "10.0.1.0/24"
  region        = "northamerica-northeast1"
  network       = google_compute_network.neurascale_vpc.id
 
  # Enable private Google access
  private_ip_google_access = true
 
  # Secondary range for pods and services
  secondary_ip_range {
    range_name    = "pods"
    ip_cidr_range = "10.1.0.0/16"
  }
 
  secondary_ip_range {
    range_name    = "services"
    ip_cidr_range = "10.2.0.0/16"
  }
}
 
# Firewall rules
resource "google_compute_firewall" "allow_internal" {
  name    = "neurascale-allow-internal"
  network = google_compute_network.neurascale_vpc.name
 
  allow {
    protocol = "tcp"
    ports    = ["0-65535"]
  }
 
  source_ranges = ["10.0.0.0/8"]
}
 
resource "google_compute_firewall" "allow_https" {
  name    = "neurascale-allow-https"
  network = google_compute_network.neurascale_vpc.name
 
  allow {
    protocol = "tcp"
    ports    = ["443"]
  }
 
  source_ranges = ["0.0.0.0/0"]
  target_tags   = ["https-server"]
}
 
# Deny all other inbound traffic
resource "google_compute_firewall" "deny_all" {
  name    = "neurascale-deny-all"
  network = google_compute_network.neurascale_vpc.name
 
  deny {
    protocol = "all"
  }
 
  source_ranges = ["0.0.0.0/0"]
  priority      = 1000
}

# DDoS protection and Web Application Firewall
resource "google_compute_security_policy" "neurascale_policy" {
  name = "neurascale-security-policy"
 
  # Rate limiting rule
  rule {
    action   = "throttle"
    priority = "1000"
 
    match {
      versioned_expr = "SRC_IPS_V1"
      config {
        src_ip_ranges = ["*"]
      }
    }
 
    rate_limit_options {
      conform_action = "allow"
      exceed_action  = "deny(429)"
 
      rate_limit_threshold {
        count        = 100
        interval_sec = 60
      }
    }
  }
 
  # Block known bad IPs
  rule {
    action   = "deny(403)"
    priority = "2000"
 
    match {
      versioned_expr = "SRC_IPS_V1"
      config {
        src_ip_ranges = [
          "192.0.2.0/24",  # Example bad IP range
        ]
      }
    }
  }
 
  # OWASP ModSecurity Core Rule Set
  rule {
    action   = "deny(403)"
    priority = "3000"
 
    match {
      expr {
        expression = "evaluatePreconfiguredExpr('xss-stable')"
      }
    }
  }
 
  rule {
    action   = "deny(403)"
    priority = "3001"
 
    match {
      expr {
        expression = "evaluatePreconfiguredExpr('sqli-stable')"
      }
    }
  }
 
  # Default allow rule
  rule {
    action   = "allow"
    priority = "2147483647"
 
    match {
      versioned_expr = "SRC_IPS_V1"
      config {
        src_ip_ranges = ["*"]
      }
    }
  }
}

class ComplianceReporting:
    """Unified compliance reporting for HIPAA and SOC 2"""
 
    def generate_compliance_report(self, report_type: str, period: str):
        report = {
            'generated_at': datetime.utcnow(),
            'period': period,
            'type': report_type
        }
 
        if report_type == 'HIPAA_AUDIT':
            report['sections'] = [
                self.generate_phi_access_report(),
                self.generate_breach_assessment(),
                self.generate_baa_status(),
                self.generate_risk_analysis()
            ]
        elif report_type == 'SOC2_ATTESTATION':
            report['sections'] = [
                self.generate_control_effectiveness(),
                self.generate_exception_report(),
                self.generate_change_management(),
                self.generate_availability_metrics()
            ]
        elif report_type == 'INTEGRATED_COMPLIANCE':
            # Combined report for healthcare clients
            report['sections'] = [
                self.generate_unified_controls(),
                self.generate_combined_risks(),
                self.generate_compliance_metrics(),
                self.generate_attestation_summary()
            ]
 
        return report
 
    def generate_unified_controls(self):
        """Show how controls satisfy both frameworks"""
        controls = []
 
        # Example: Access control satisfies both
        controls.append({
            'control_id': 'AC-001',
            'description': 'Role-based access control with MFA',
            'hipaa_requirements': ['§164.308(a)(4)', '§164.312(a)(1)'],
            'soc2_criteria': ['CC6.1', 'CC6.2', 'CC6.3'],
            'effectiveness': 'Operating effectively',
            'last_tested': '2024-01-15',
            'evidence': ['access_logs', 'user_reviews', 'mfa_reports']
        })
 
        return controls

# Automated data lifecycle management
class DataLifecycleManager:
    def __init__(self):
        self.retention_policies = {
            'clinical_data': timedelta(days=2555),  # 7 years for HIPAA
            'research_data': timedelta(days=3653),  # 10 years for research
            'audit_logs': timedelta(days=2555),     # 7 years for compliance
            'system_logs': timedelta(days=90),      # 90 days for operations
            'backup_data': timedelta(days=2555),    # 7 years for recovery
        }
 
    def apply_retention_policy(self):
        """Apply retention policies across all data types"""
 
        for data_type, retention_period in self.retention_policies.items():
            cutoff_date = datetime.utcnow() - retention_period
 
            if data_type == 'clinical_data':
                self.archive_clinical_data(cutoff_date)
            elif data_type == 'audit_logs':
                self.archive_audit_logs(cutoff_date)
            # ... handle other data types
 
    def secure_delete(self, file_path: str):
        """Securely delete files using DoD 5220.22-M standard"""
 
        # Three-pass overwrite
        with open(file_path, 'r+b') as f:
            file_size = f.seek(0, 2)
            f.seek(0)
 
            # Pass 1: Write random data
            f.write(os.urandom(file_size))
            f.flush()
            os.fsync(f.fileno())
 
            # Pass 2: Write complement
            f.seek(0)
            f.write(bytes([0xFF] * file_size))
            f.flush()
            os.fsync(f.fileno())
 
            # Pass 3: Write zeros
            f.seek(0)
            f.write(bytes([0x00] * file_size))
            f.flush()
            os.fsync(f.fileno())
 
        # Remove file
        os.remove(file_path)
 
        # Log destruction
        audit_log.log_data_destruction(
            file_path=file_path,
            destruction_method="DoD_5220.22-M",
            timestamp=datetime.utcnow()
        )

# Security monitoring and alerting
class SecurityMonitor:
    def __init__(self):
        self.alert_thresholds = {
            'failed_logins': 5,
            'suspicious_api_calls': 100,
            'data_export_volume': 1000,  # MB
            'unusual_access_patterns': 10
        }
 
    def monitor_failed_logins(self):
        """Monitor for brute force attacks"""
 
        # Query recent failed logins
        recent_failures = db.session.query(AuditLog).filter(
            AuditLog.event_type == 'LOGIN_FAILED',
            AuditLog.timestamp > datetime.utcnow() - timedelta(minutes=15)
        ).all()
 
        # Group by IP address
        failures_by_ip = {}
        for failure in recent_failures:
            ip = failure.ip_address
            failures_by_ip[ip] = failures_by_ip.get(ip, 0) + 1
 
        # Alert on threshold breach
        for ip, count in failures_by_ip.items():
            if count >= self.alert_thresholds['failed_logins']:
                self.send_security_alert(
                    alert_type='BRUTE_FORCE_DETECTED',
                    details={
                        'ip_address': ip,
                        'failed_attempts': count,
                        'time_window': '15 minutes'
                    },
                    severity='HIGH'
                )
 
                # Temporarily block IP
                self.block_ip_address(ip, duration_minutes=60)
 
    def detect_anomalous_access(self, user_id: str):
        """Detect unusual access patterns"""
 
        # Get user's typical access patterns
        user_profile = self.get_user_access_profile(user_id)
 
        # Current session analysis
        current_session = self.get_current_session(user_id)
 
        anomalies = []
 
        # Check for unusual times
        if self.is_unusual_time(current_session.login_time, user_profile):
            anomalies.append('unusual_time')
 
        # Check for unusual location
        if self.is_unusual_location(current_session.ip_address, user_profile):
            anomalies.append('unusual_location')
 
        # Check for unusual data access volume
        if self.is_unusual_data_access(current_session, user_profile):
            anomalies.append('unusual_data_volume')
 
        if len(anomalies) >= 2:  # Multiple anomalies = high risk
            self.send_security_alert(
                alert_type='ANOMALOUS_ACCESS_DETECTED',
                details={
                    'user_id': user_id,
                    'anomalies': anomalies,
                    'session_id': current_session.id
                },
                severity='MEDIUM'
            )
 
            # Require additional authentication
            self.require_step_up_auth(user_id)
 
    def send_security_alert(self, alert_type: str, details: dict, severity: str):
        """Send security alerts to SOC team"""
 
        alert = {
            'timestamp': datetime.utcnow().isoformat(),
            'alert_type': alert_type,
            'severity': severity,
            'details': details,
            'source': 'neurascale-security-monitor'
        }
 
        # Send to Slack/PagerDuty
        if severity in ['HIGH', 'CRITICAL']:
            self.send_pagerduty_alert(alert)
        else:
            self.send_slack_alert(alert)
 
        # Log to security incident system
        self.log_security_incident(alert)

# Input validation and sanitization
from marshmallow import Schema, fields, validate, ValidationError
import bleach
 
class NeuralDataSchema(Schema):
    device_id = fields.Str(
        required=True,
        validate=validate.Regexp(r'^[a-zA-Z0-9_-]+$'),
        error_messages={'invalid': 'Device ID contains invalid characters'}
    )
 
    sample_rate = fields.Int(
        required=True,
        validate=validate.Range(min=1, max=10000),
        error_messages={'invalid': 'Sample rate must be between 1 and 10000 Hz'}
    )
 
    channels = fields.List(
        fields.Int(validate=validate.Range(min=1, max=256)),
        validate=validate.Length(min=1, max=64),
        required=True
    )
 
    metadata = fields.Dict(
        keys=fields.Str(validate=validate.Length(max=50)),
        values=fields.Str(validate=validate.Length(max=500))
    )
 
def validate_and_sanitize_input(data: dict) -> dict:
    """Validate and sanitize all user inputs"""
 
    schema = NeuralDataSchema()
 
    try:
        # Validate structure and types
        validated_data = schema.load(data)
 
        # Sanitize string fields
        if 'metadata' in validated_data:
            for key, value in validated_data['metadata'].items():
                validated_data['metadata'][key] = bleach.clean(
                    value,
                    tags=[],  # No HTML tags allowed
                    strip=True
                )
 
        return validated_data
 
    except ValidationError as err:
        raise ValueError(f"Input validation failed: {err.messages}")
 
# SQL injection prevention
def get_user_sessions(user_id: str, start_date: datetime, end_date: datetime):
    """Safe database query using parameterized statements"""
 
    # Input validation
    if not isinstance(user_id, str) or not user_id.strip():
        raise ValueError("Invalid user_id")
 
    if start_date >= end_date:
        raise ValueError("start_date must be before end_date")
 
    # Parameterized query (prevents SQL injection)
    query = """
        SELECT id, name, start_time, end_time, status
        FROM sessions
        WHERE user_id = %s
          AND start_time >= %s
          AND start_time <= %s
        ORDER BY start_time DESC
        LIMIT 100
    """
 
    return db.session.execute(
        text(query),
        (user_id, start_date, end_date)
    ).fetchall()
 
# XSS prevention
def render_user_content(content: str) -> str:
    """Safely render user-generated content"""
 
    # Whitelist allowed HTML tags and attributes
    allowed_tags = ['p', 'br', 'strong', 'em', 'ul', 'ol', 'li']
    allowed_attributes = {}
 
    cleaned_content = bleach.clean(
        content,
        tags=allowed_tags,
        attributes=allowed_attributes,
        strip=True
    )
 
    return cleaned_content

# Security testing framework
import pytest
import requests
from selenium import webdriver
from zap import ZAPv2
 
class SecurityTestSuite:
    def __init__(self):
        self.base_url = "https://staging-api.neurascale.io"
        self.zap = ZAPv2()
 
    def test_authentication_bypass(self):
        """Test for authentication bypass vulnerabilities"""
 
        protected_endpoints = [
            '/api/v1/devices',
            '/api/v1/sessions',
            '/api/v1/users/profile'
        ]
 
        for endpoint in protected_endpoints:
            # Test without token
            response = requests.get(f"{self.base_url}{endpoint}")
            assert response.status_code == 401, f"Endpoint {endpoint} not properly protected"
 
            # Test with invalid token
            headers = {'Authorization': 'Bearer invalid_token'}
            response = requests.get(f"{self.base_url}{endpoint}", headers=headers)
            assert response.status_code == 401, f"Endpoint {endpoint} accepts invalid tokens"
 
    def test_sql_injection(self):
        """Test for SQL injection vulnerabilities"""
 
        # Common SQL injection payloads
        payloads = [
            "'; DROP TABLE users; --",
            "' OR '1'='1",
            "' UNION SELECT * FROM users --",
            "admin'/*",
            "'; EXEC xp_cmdshell('dir'); --"
        ]
 
        for payload in payloads:
            # Test in various parameters
            test_data = {
                'device_id': payload,
                'user_id': payload,
                'session_name': payload
            }
 
            response = requests.post(
                f"{self.base_url}/api/v1/sessions",
                json=test_data,
                headers={'Authorization': 'Bearer valid_test_token'}
            )
 
            # Should return 400 (validation error) not 500 (database error)
            assert response.status_code != 500, f"Potential SQL injection with payload: {payload}"
 
    def test_xss_prevention(self):
        """Test for XSS vulnerabilities"""
 
        xss_payloads = [
            "<script>alert('XSS')</script>",
            "<img src=x onerror=alert('XSS')>",
            "javascript:alert('XSS')",
            "<svg onload=alert('XSS')>",
            "';alert('XSS');//"
        ]
 
        for payload in xss_payloads:
            # Test in session metadata
            test_data = {
                'name': f"Test Session {payload}",
                'description': f"Description with {payload}",
                'metadata': {'notes': payload}
            }
 
            response = requests.post(
                f"{self.base_url}/api/v1/sessions",
                json=test_data,
                headers={'Authorization': 'Bearer valid_test_token'}
            )
 
            if response.status_code == 201:
                # Verify payload was sanitized
                session_data = response.json()
                assert payload not in str(session_data), f"XSS payload not sanitized: {payload}"
 
    def test_rate_limiting(self):
        """Test rate limiting implementation"""
 
        # Rapid requests to test rate limiting
        responses = []
        for i in range(150):  # Exceed 100 req/min limit
            response = requests.get(f"{self.base_url}/api/v1/status")
            responses.append(response.status_code)
 
        # Should see 429 (Too Many Requests) responses
        assert 429 in responses, "Rate limiting not properly implemented"
 
    def run_zap_security_scan(self):
        """Run OWASP ZAP security scan"""
 
        # Start ZAP daemon
        self.zap.core.new_session()
 
        # Spider the application
        self.zap.spider.scan(self.base_url)
 
        # Wait for spider to complete
        while int(self.zap.spider.status()) < 100:
            time.sleep(1)
 
        # Active security scan
        self.zap.ascan.scan(self.base_url)
 
        # Wait for scan to complete
        while int(self.zap.ascan.status()) < 100:
            time.sleep(5)
 
        # Get results
        alerts = self.zap.core.alerts()
 
        # Filter high-risk issues
        high_risk_alerts = [
            alert for alert in alerts
            if alert['risk'] in ['High', 'Critical']
        ]
 
        assert len(high_risk_alerts) == 0, f"High-risk security issues found: {high_risk_alerts}"

# Cloud Monitoring alerting policies
resource "google_monitoring_alert_policy" "failed_login_attempts" {
  display_name = "High Failed Login Attempts"
  combiner     = "OR"
 
  conditions {
    display_name = "Failed login rate"
 
    condition_threshold {
      filter          = "resource.type=\"cloud_run_revision\" AND jsonPayload.event_type=\"LOGIN_FAILED\""
      duration        = "300s"
      comparison      = "COMPARISON_GT"
      threshold_value = 10
 
      aggregations {
        alignment_period   = "60s"
        per_series_aligner = "ALIGN_RATE"
      }
    }
  }
 
  notification_channels = [
    google_monitoring_notification_channel.pagerduty.name
  ]
 
  alert_strategy {
    auto_close = "1800s"
  }
}
 
resource "google_monitoring_alert_policy" "data_exfiltration" {
  display_name = "Unusual Data Export Volume"
  combiner     = "OR"
 
  conditions {
    display_name = "High data export rate"
 
    condition_threshold {
      filter          = "resource.type=\"cloud_run_revision\" AND jsonPayload.event_type=\"DATA_EXPORT\""
      duration        = "900s"
      comparison      = "COMPARISON_GT"
      threshold_value = 1000000000  # 1GB
 
      aggregations {
        alignment_period   = "300s"
        per_series_aligner = "ALIGN_SUM"
      }
    }
  }
 
  notification_channels = [
    google_monitoring_notification_channel.security_team.name
  ]
}