Capability Closure Implementation Plan

1
2
3
4
5
def test_oidc_provider_rejects_invalid_signature():
    # given a JWT signed with a different key
    # when validate_token is called
    # then it raises 401 / returns False
    ...

1
2
3
4
# services/workbench-bff/src/api/jwks_cache.py
class JwksCache:
    def get_signing_key(self, jwks_url: str, kid: str) -> dict:
        ...

1
2
3
4
5
6
7
8
9
def validate_auth_config():
    """Fail-fast validation on app startup"""
    if os.getenv("ENVIRONMENT") == "production":
        if os.getenv("AUTH_PROVIDER") != "oidc":
            raise ValueError("AUTH_PROVIDER must be 'oidc' in production")
        required_vars = ["OIDC_ISSUER_URL", "OIDC_CLIENT_ID", "OIDC_AUDIENCE"]
        missing = [v for v in required_vars if not os.getenv(v)]
        if missing:
            raise ValueError(f"Missing required OIDC variables: {missing}")

1
2
3
4
def test_remediation_engine_creates_pr_when_enabled():
    # mock GitHub API client
    # expect PR URL returned
    ...

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
import os
import time
import hashlib
from typing import Optional
import httpx
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type

class GitHubAdapter:
    """Production-ready GitHub PR creation adapter"""

    def __init__(self):
        # Fetch token from secrets manager (not plain env)
        self.token = self._get_github_token()
        self.repo = os.getenv('GITHUB_REMEDIATION_REPO')
        self.base_branch = os.getenv('GITHUB_BASE_BRANCH', 'main')

        if not self.repo:
            raise ValueError("GITHUB_REMEDIATION_REPO must be set")

        self.client = httpx.AsyncClient(
            base_url='https://api.github.com',
            headers={
                'Authorization': f'Bearer {self.token}',
                'Accept': 'application/vnd.github.v3+json',
                'X-GitHub-Api-Version': '2022-11-28'
            },
            timeout=30.0
        )

    def _get_github_token(self) -> str:
        """Fetch GitHub token from secrets manager

        Required permissions: repo (read/write), pull_requests (write)
        """
        # TODO: Integrate with AWS Secrets Manager / Vault
        # For now, check env but document this is temporary
        token = os.getenv('GITHUB_TOKEN')
        if not token:
            raise ValueError("GITHUB_TOKEN must be set (temporary - migrate to secrets manager)")
        return token

    @retry(
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=1, max=10),
        retry=retry_if_exception_type((httpx.HTTPStatusError, httpx.NetworkError)),
        reraise=True
    )
    async def create_pr(
        self,
        title: str,
        body: str,
        branch: str,
        files: dict[str, str]
    ) -> Optional[str]:
        """Create PR with exponential backoff and idempotency

        Returns:
            PR URL if created, None if already exists
        """
        try:
            # Check idempotency: search for existing PR/branch
            existing_pr = await self._find_existing_pr(branch)
            if existing_pr:
                self._log_audit('pr_skip', {
                    'branch': branch,
                    'reason': 'already_exists',
                    'pr_url': existing_pr
                })
                return existing_pr

            # Create branch and commit files
            await self._create_branch_with_files(branch, files)

            # Create pull request
            pr_response = await self.client.post(
                f'/repos/{self.repo}/pulls',
                json={
                    'title': title,
                    'body': body,
                    'head': branch,
                    'base': self.base_branch
                }
            )
            pr_response.raise_for_status()

            pr_url = pr_response.json()['html_url']

            self._log_audit('pr_created', {
                'branch': branch,
                'pr_url': pr_url,
                'timestamp': time.time()
            })

            return pr_url

        except httpx.HTTPStatusError as e:
            # Translate GitHub errors into actionable responses
            if e.response.status_code == 403:
                self._log_audit('pr_error', {
                    'error': 'permission_denied',
                    'message': 'Check GitHub token permissions (repo, pull_requests)',
                    'response': e.response.text
                })
                raise PermissionError("GitHub token lacks required permissions")

            elif e.response.status_code == 404:
                self._log_audit('pr_error', {
                    'error': 'repo_not_found',
                    'repo': self.repo
                })
                raise ValueError(f"Repository not found: {self.repo}")

            elif e.response.status_code == 429:
                self._log_audit('pr_error', {
                    'error': 'rate_limit',
                    'retry_after': e.response.headers.get('Retry-After')
                })
                # Retry will handle this
                raise

            else:
                self._log_audit('pr_error', {
                    'error': 'github_api_error',
                    'status': e.response.status_code,
                    'message': e.response.text
                })
                raise

        except httpx.NetworkError as e:
            self._log_audit('pr_error', {
                'error': 'network_error',
                'message': str(e)
            })
            # Retry will handle this
            raise

    async def _find_existing_pr(self, branch: str) -> Optional[str]:
        """Check if PR already exists for branch"""
        try:
            response = await self.client.get(
                f'/repos/{self.repo}/pulls',
                params={'head': f'{self.repo.split("/")[0]}:{branch}', 'state': 'open'}
            )
            response.raise_for_status()

            prs = response.json()
            if prs:
                return prs[0]['html_url']

            return None

        except httpx.HTTPStatusError:
            # If search fails, proceed with creation attempt
            return None

    def _log_audit(self, event_type: str, details: dict):
        """Structured audit logging for all PR operations"""
        import logging
        import json

        logger = logging.getLogger('github_adapter')
        logger.info(json.dumps({
            'event': event_type,
            'service': 'remediation_engine',
            'timestamp': time.time(),
            **details
        }))

# Update RemediationEngine to return pr_url
class RemediationEngine:
    def __init__(self):
        self.github = GitHubAdapter()

    async def create_remediation_pr(self, drift_data: dict) -> dict:
        """Create PR and return result with URL"""
        try:
            pr_url = await self.github.create_pr(
                title=f"Auto-remediation: {drift_data['title']}",
                body=self._generate_pr_body(drift_data),
                branch=f"remediation/{drift_data['id']}",
                files=drift_data['files']
            )

            return {
                'status': 'success',
                'pr_url': pr_url,
                'pr_created': pr_url is not None
            }

        except Exception as e:
            return {
                'status': 'error',
                'error': str(e),
                'pr_url': None
            }

1
2
3
4
5
6
7
8
9
10
11
12
13
14
@router.post('/drift/remediate')
async def create_remediation(drift_id: str):
    """Trigger drift remediation with PR creation"""
    # ... existing drift detection logic ...

    result = await remediation_engine.create_remediation_pr(drift_data)

    # Ensure pr_url is included in response
    return {
        'drift_id': drift_id,
        'remediation_status': result['status'],
        'pr_url': result.get('pr_url'),  # May be None if error
        'pr_created': result.get('pr_created', False)
    }

1
2
def test_gate_strict_mode_fails_without_api():
    ...

1
2
3
4
5
async def test_sparql_reasoning_profile_roundtrip():
    # insert explicit graph
    # query with reasoning_profile=rdfs
    # expect inferred metadata + named graph
    ...

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
import yaml
from pathlib import Path
from typing import Dict, Any

class PerformanceValidator:
    def __init__(self, config_path: str):
        with open(config_path) as f:
            self.config = yaml.safe_load(f)

    def validate_slo(self, service: str, metrics: Dict[str, float]) -> bool:
        """Check if metrics meet SLO thresholds"""
        slo = self.config['slos'][service]
        violations = []

        if metrics['p95_latency_ms'] > slo['p95_latency_ms']:
            violations.append(f"P95 latency {metrics['p95_latency_ms']}ms > {slo['p95_latency_ms']}ms")

        if metrics['throughput_rps'] < slo['min_throughput_rps']:
            violations.append(f"Throughput {metrics['throughput_rps']} < {slo['min_throughput_rps']} rps")

        if metrics['error_rate'] > slo['max_error_rate']:
            violations.append(f"Error rate {metrics['error_rate']} > {slo['max_error_rate']}")

        if violations:
            raise AssertionError(f"SLO violations for {service}: " + "; ".join(violations))

        return True

    def compare_baseline(self, service: str, current: Dict[str, float]) -> bool:
        """Compare current metrics against stored baseline"""
        baseline_path = Path(self.config['baseline_storage']['path']) / f"{service}.json"

        if not baseline_path.exists():
            # First run, store as baseline
            self.store_baseline(service, current)
            return True

        with open(baseline_path) as f:
            baseline = json.load(f)

        thresholds = self.config['regression_thresholds']

        # Check p95 regression
        p95_increase = ((current['p95_latency_ms'] - baseline['p95_latency_ms'])
                        / baseline['p95_latency_ms'] * 100)
        if p95_increase > thresholds['p95_latency_increase_percent']:
            raise AssertionError(f"P95 latency regression: {p95_increase:.1f}% increase")

        # Check throughput regression
        throughput_decrease = ((baseline['throughput_rps'] - current['throughput_rps'])
                              / baseline['throughput_rps'] * 100)
        if throughput_decrease > thresholds['throughput_decrease_percent']:
            raise AssertionError(f"Throughput regression: {throughput_decrease:.1f}% decrease")

        return True

1
2
def test_chaos_runner_loads_scenarios():
    ...

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
import os
import sys
import yaml
from pathlib import Path

class ChaosRunner:
    def __init__(self, config_path: str = "tests/chaos/config.yaml"):
        with open(config_path) as f:
            self.config = yaml.safe_load(f)

        self._safety_check()

    def _safety_check(self):
        """Abort if running in unsafe environment"""
        if self.config['safety']['environment_check']:
            env = os.getenv('ENVIRONMENT', 'development')
            if env == 'production':
                raise RuntimeError("CHAOS ABORT: Cannot run chaos tests in production environment")

        if self.config['safety']['require_explicit_enable']:
            if not os.getenv('CHAOS_ENABLED') and '--enable-chaos' not in sys.argv:
                raise RuntimeError("CHAOS ABORT: Must set CHAOS_ENABLED=true or pass --enable-chaos flag")

    def run_scenario(self, scenario_name: str):
        """Execute chaos scenario with automated cleanup"""
        scenario = self.config['scenarios'][scenario_name]

        try:
            # Verify steady state
            self._verify_steady_state(scenario['steady_state'])

            # Execute chaos actions
            for action in scenario['actions']:
                self._execute_action(action)

            # Run validation probes
            results = self._validate(scenario['validation'])

            # Emit structured results
            self._emit_results(scenario_name, results)

        finally:
            # ALWAYS run cleanup
            self._cleanup(scenario['cleanup'])

    def _cleanup(self, cleanup_steps):
        """Guaranteed cleanup regardless of test outcome"""
        for step in cleanup_steps:
            try:
                self._execute_cleanup_step(step)
            except Exception as e:
                print(f"CLEANUP ERROR: {step}: {e}", file=sys.stderr)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
# tests/chaos/scenarios/nats_partition_mesh.py
from ..probes import nats_probe, http_probe

def execute(config):
    # Emit structured probe results
    probe_results = {
        'pre_chaos': {
            'http': http_probe.check_health('http://localhost:8000/health'),
            'nats': nats_probe.check_stream('events')
        },
        'during_chaos': {},
        'post_chaos': {}
    }

    # Apply chaos...
    # Collect probe results...

    return probe_results