Defense in Depth: Lessons from Cloudflare

1
class FeatureFileValidator:
2
    """Validate feature configuration files before deployment."""
3

4
    MAX_FEATURES = 200
5
    MAX_FILE_SIZE = 10 * 1024 * 1024  # 10MB
6

7
    def validate(self, file_contents: bytes, file_path: str) -> ValidatedFeatures:
8
        """
9
        Validate feature file with multiple checks.
10
        Returns validated features or raises ValidationError.
11
        """
12
        # Check 1: File size
13
        if len(file_contents) > self.MAX_FILE_SIZE:
14
            raise ValidationError(
15
                f"File too large: {len(file_contents)} bytes > {self.MAX_FILE_SIZE}",
16
                file=file_path
17
            )
18

19
        # Check 2: Parse the file
20
        try:
21
            features = parse_features(file_contents)
22
        except ParseError as e:
23
            raise ValidationError(f"Failed to parse file: {e}", file=file_path)
24

25
        # Check 3: Feature count
26
        if len(features) > self.MAX_FEATURES:
27
            raise ValidationError(
28
                f"Too many features: {len(features)} > {self.MAX_FEATURES}",
29
                file=file_path,
30
                feature_count=len(features)
31
            )
32

33
        if len(features) == 0:
34
            raise ValidationError("No features found", file=file_path)
35

36
        # Check 4: Duplicate detection
37
        feature_names = [f.name for f in features]
38
        duplicates = find_duplicates(feature_names)
39
        if duplicates:
40
            raise ValidationError(
41
                f"Duplicate features detected: {duplicates}",
42
                file=file_path
43
            )
44

45
        # Check 5: Schema validation
46
        for feature in features:
47
            if not self._validate_feature_schema(feature):
48
                raise ValidationError(
49
                    f"Invalid feature schema: {feature.name}",
50
                    file=file_path
51
                )
52

53
        # Check 6: Sudden size changes (anomaly detection)
54
        previous_count = self._get_previous_feature_count()
55
        if previous_count and len(features) > previous_count * 1.5:
56
            raise ValidationError(
57
                f"Suspicious feature count increase: {previous_count} → {len(features)}",
58
                file=file_path,
59
                alert_ops_team=True
60
            )
61

62
        return ValidatedFeatures(features, file_path)
63

64
    def _validate_feature_schema(self, feature: Feature) -> bool:
65
        """Validate individual feature schema."""
66
        required_fields = ['name', 'type', 'metadata']
67
        return all(hasattr(feature, field) for field in required_fields)
68

69
# Usage in deployment pipeline
70
def deploy_feature_file(file_path: str):
71
    """Deploy feature file with validation."""
72
    with open(file_path, 'rb') as f:
73
        contents = f.read()
74

75
    validator = FeatureFileValidator()
76

77
    try:
78
        validated = validator.validate(contents, file_path)
79
    except ValidationError as e:
80
        log_error(f"Feature file validation failed: {e}")
81
        alert_ops_team(e)
82
        # CRITICAL: Don't deploy invalid file!
83
        return False
84

85
    # Deploy only if validation passed
86
    deploy_to_network(validated.features)
87
    return True

1
fn process_request(request: HttpRequest) -> Result<HttpResponse, Error> {
2
    let features = load_bot_features()?.unwrap();  // Panic on error!
3
    let bot_score = calculate_bot_score(&request, &features);
4
    Ok(build_response(bot_score))
5
}

1
fn process_request(request: HttpRequest) -> Result<HttpResponse, Error> {
2
    let features = match load_bot_features() {
3
        Ok(Some(f)) => f,
4
        Ok(None) | Err(_) => {
5
            // Graceful degradation strategy
6
            log_error!("Bot features unavailable, using fallback");
7
            metrics::increment("bot_management.degraded_mode");
8

9
            // Option 1: Use cached features
10
            if let Some(cached) = get_cached_features() {
11
                cached
12
            } else {
13
                // Option 2: Disable bot scoring, allow traffic
14
                return Ok(build_response_without_bot_scoring());
15
            }
16
        }
17
    };
18

19
    let bot_score = calculate_bot_score(&request, &features);
20
    Ok(build_response(bot_score))
21
}
22

23
fn get_cached_features() -> Option<Features> {
24
    """Return last known-good features from cache."""
25
    FEATURE_CACHE.read()
26
        .ok()
27
        .and_then(|cache| cache.get_last_valid())
28
}

1
class BotManagementModule:
2
    def __init__(self):
3
        self.feature_cache = FeatureCache(ttl=3600)  # 1 hour cache
4
        self.degraded_mode = False
5

6
    def process_request(self, request):
7
        """Process request with automatic fallback."""
8
        try:
9
            features = self.load_features()
10
        except TooManyFeaturesError as e:
11
            # Degraded mode: use cached features
12
            log_error(f"Feature load failed: {e}, entering degraded mode")
13
            self.degraded_mode = True
14
            features = self.feature_cache.get_last_valid()
15

16
            if features is None:
17
                # Ultimate fallback: disable bot scoring
18
                log_critical("No cached features, disabling bot module")
19
                return self.process_without_bot_scoring(request)
20

21
        bot_score = self.calculate_score(request, features)
22
        return self.build_response(request, bot_score)

1
class FeatureFlagSystem:
2
    """Centralized feature flag management with kill switches."""
3

4
    def __init__(self):
5
        self.flags = self._load_flags_from_config()
6
        self.kill_switches = KillSwitchManager()
7

8
    def is_enabled(self, feature_name: str, request_context: dict = None) -> bool:
9
        """Check if feature is enabled, respecting kill switches."""
10

11
        # Check 1: Global kill switch (highest priority)
12
        if self.kill_switches.is_active(f"{feature_name}_global_disable"):
13
            log_info(f"Feature {feature_name} disabled by global kill switch")
14
            metrics.increment(f"{feature_name}.killed_globally")
15
            return False
16

17
        # Check 2: Datacenter-specific kill switch
18
        if request_context:
19
            dc = request_context.get('datacenter')
20
            if self.kill_switches.is_active(f"{feature_name}_disable_{dc}"):
21
                return False
22

23
        # Check 3: Normal feature flag
24
        flag = self.flags.get(feature_name)
25
        if not flag or not flag.enabled:
26
            return False
27

28
        # Check 4: Automatic circuit breaker
29
        if self._is_circuit_broken(feature_name):
30
            log_warning(f"Feature {feature_name} circuit broken due to errors")
31
            return False
32

33
        return True
34

35
    def _is_circuit_broken(self, feature_name: str) -> bool:
36
        """Automatic circuit breaker based on error rates."""
37
        error_count = metrics.get(f"{feature_name}.errors.count.5min")
38
        error_rate = metrics.get(f"{feature_name}.errors.rate.5min")
39

40
        # Break circuit if error rate > 50% or errors > 1000/min
41
        if error_rate and error_rate > 0.5:
42
            return True
43
        if error_count and error_count > 1000:
44
            return True
45

46
        return False
47

48
# Usage in Bot Management
49
def process_request(request):
50
    if not feature_flags.is_enabled('bot_management', request.context):
51
        # Feature disabled, skip bot management
52
        return process_without_bot_scoring(request)
53

54
    # Normal bot management processing
55
    return process_with_bot_management(request)

1
class AdaptiveObservability:
2
    """Observability system that adapts to error rates."""
3

4
    def __init__(self):
5
        self.sample_rate = 1.0  # 100% sampling normally
6
        self.error_rate_threshold = 100  # errors/sec
7
        self.high_volume_sample_rate = 0.01  # 1% during incidents
8

9
    def record_error(self, error: Exception, context: dict):
10
        """Record error with adaptive sampling."""
11
        current_error_rate = self._get_error_rate()
12

13
        # Adapt sample rate based on error volume
14
        if current_error_rate > self.error_rate_threshold:
15
            # High error rate: reduce sampling to save CPU
16
            self.sample_rate = self.high_volume_sample_rate
17
            metrics.increment("observability.high_error_mode")
18
        else:
19
            # Normal error rate: full sampling
20
            self.sample_rate = 1.0
21

22
        # Sample decision
23
        if not self._should_sample():
24
            # Still count the error, just don't enhance it
25
            metrics.increment(f"errors.{error.__class__.__name__}")
26
            return
27

28
        # Full error enhancement (expensive)
29
        enhanced_error = self._enhance_error(error, context)
30
        self._log_to_storage(enhanced_error)
31

32
    def _should_sample(self) -> bool:
33
        """Probabilistic sampling."""
34
        return random.random() < self.sample_rate
35

36
    def _enhance_error(self, error: Exception, context: dict) -> dict:
37
        """Enhance error with stack trace, context, etc. EXPENSIVE."""
38
        return {
39
            'error': str(error),
40
            'type': error.__class__.__name__,
41
            'stack_trace': traceback.format_exc(),  # Expensive!
42
            'context': context,
43
            'timestamp': time.time(),
44
            'request_id': context.get('request_id'),
45
            # ... more expensive debugging data
46
        }

1
from hypothesis import given, strategies as st
2
import pytest
3

4
class TestFeatureFileGeneration:
5
    @given(
6
        feature_count=st.integers(min_value=0, max_value=500),
7
        database_count=st.integers(min_value=1, max_value=5)
8
    )
9
    def test_feature_file_handles_any_count_gracefully(
10
        self, feature_count, database_count
11
    ):
12
        """
13
        Test that feature file generation handles any number of
14
        features and databases gracefully.
15
        """
16
        # Simulate varying database visibility
17
        mock_query_result = generate_mock_columns(
18
            feature_count, database_count
19
        )
20

21
        result = generate_feature_file(mock_query_result)
22

23
        if feature_count > MAX_FEATURES:
24
            # Should return error, not crash
25
            assert result.is_err()
26
            assert "too many features" in result.error_message().lower()
27
        else:
28
            assert result.is_ok()
29
            assert len(result.features) <= MAX_FEATURES
30

31
    def test_query_filters_database_explicitly(self):
32
        """
33
        CRITICAL: Ensure query explicitly filters by database.
34

35
        This test would have caught the Cloudflare bug!
36
        """
37
        query = get_feature_discovery_query()
38

39
        # Assertion: Query must contain database filter
40
        assert "database = 'default'" in query.lower(), \
41
            "Query must explicitly filter by database name!"

1
def get_feature_discovery_query() -> str:
2
    """
3
    Get SQL query to discover available features.
4

5
    ASSUMPTIONS (CRITICAL - TEST THESE!):
6
    1. Only returns columns from 'default' database
7
    2. Returns at most MAX_FEATURES results
8
    3. No duplicate feature names in result
9
    4. 'http_requests_features' table exists
10

11
    QUERY DEPENDENCIES:
12
    - ClickHouse permissions: assumes 'default' database is visible
13
    - Schema: assumes 'system.columns' table exists
14

15
    If permissions change, THIS QUERY MAY BREAK!
16
    """
17
    return """
18
        SELECT name, type
19
        FROM system.columns
20
        WHERE database = 'default'  -- CRITICAL: Explicit filter!
21
          AND table = 'http_requests_features'
22
        ORDER BY name
23
    """

1
class AssumptionValidator:
2
    """Validate runtime assumptions that code depends on."""
3

4
    @staticmethod
5
    def validate_feature_query_assumptions(query_result: List[dict]):
6
        """
7
        Validate assumptions about feature query results.
8
        Fail fast if assumptions are violated.
9
        """
10
        # Assumption 1: All rows are from 'default' database
11
        databases = {row['database'] for row in query_result}
12
        assert databases == {'default'}, \
13
            f"Expected only 'default' database, got: {databases}"
14

15
        # Assumption 2: No duplicate feature names
16
        names = [row['name'] for row in query_result]
17
        duplicates = find_duplicates(names)
18
        assert not duplicates, \
19
            f"Duplicate feature names: {duplicates}"
20

21
        # Assumption 3: Feature count within limits
22
        assert len(query_result) <= MAX_FEATURES, \
23
            f"Feature count {len(query_result)} exceeds max {MAX_FEATURES}"

1
def generate_feature_file():
2
    query = get_feature_discovery_query()
3
    results = execute_query(query)
4

5
    # Validate assumptions BEFORE processing
6
    AssumptionValidator.validate_feature_query_assumptions(results)
7

8
    return build_feature_file(results)