v0.7.2-beta: ROA Planner / ROA Preflight

README.md

@@ -43,7 +43,7 @@ Routing changes often require fast but traceable checks across multiple external
 
 ## Current Alpha Status
 
-RouteForge is a **functional beta** release with production-like workflows for read-only validation and demo usage. Current release target: **v0.7.1-beta**.
+RouteForge is a **functional beta** release with production-like workflows for read-only validation and demo usage. Current release target: **v0.7.2-beta**.
 
 ## Quickstart with Docker Compose
 
@@ -276,6 +276,6 @@ In the standard Docker setup, API calls are same-origin via frontend nginx (`/ap
 - Keep `SECRET_KEY` stable; changing it invalidates existing sessions.
 
 
-## BGP Visibility Details (v0.7.1-beta)
+## BGP Visibility Details (v0.7.2-beta)
 - Read-only BGP visibility validation for prefix and optional expected origin AS.
 - Uses external RIPEstat visibility data; results are momentary snapshots and do not replace continuous monitoring.

RELEASE_NOTES.md

@@ -1,4 +1,4 @@
-## v0.7.1-beta: BGP Visibility Details
+## v0.7.2-beta: BGP Visibility Details
 
 ### Motivation
 Improve prefix visibility checks with explicit BGP origin visibility details while preserving RouteForge's strict read-only model.

ROADMAP.md

@@ -1,13 +1,13 @@
 # RouteForge Roadmap
 
 ## Current Status
-v0.7.1-beta, BGP Visibility Details completed, read-only
+v0.7.2-beta, BGP Visibility Details completed, read-only
 
-## v0.7.1-beta
+## v0.7.2-beta
 - projects/change cases
 - grouped preflight reports
 
-## v0.7.1-beta
+## v0.7.2-beta
 - bgp visibility details
 
 ## v0.7.2-beta

backend/app/api/routes_checks.py

@@ -6,12 +6,13 @@
 from app.core.audit import write_audit_log
 from app.database import get_db
 from app.models import ChangeCase, Check, Report
-from app.schemas import AsnCheckRequest, AsnRpkiBatchRequest, BgpVisibilityCheckRequest, CheckResponse, PrefixCheckRequest, PreflightCheckRequest
+from app.schemas import AsnCheckRequest, AsnRpkiBatchRequest, BgpVisibilityCheckRequest, CheckResponse, PrefixCheckRequest, PreflightCheckRequest, RoaPreflightCheckRequest
 from app.services.asn_checker import AsnChecker
 from app.services.bgp_visibility_service import BgpVisibilityService
 from app.services.prefix_checker import PrefixChecker
 from app.services.preflight_checker import PreflightChecker
 from app.services.report_renderer import render_report
+from app.services.roa_planner_service import RoaPlannerService
 from app.services.ripe_stat_client import RipeStatClient
 
 router = APIRouter(prefix="/api/check", tags=["checks"])
@@ -65,6 +66,19 @@ def check_bgp_visibility(payload: BgpVisibilityCheckRequest, db: Session = Depen
     except Exception as exc:
         raise HTTPException(status_code=500, detail=f"BGP-Visibility-Prüfung fehlgeschlagen: {exc}") from exc
 
+@router.post('/roa-preflight', response_model=CheckResponse)
+def check_roa_preflight(payload: RoaPreflightCheckRequest, db: Session = Depends(get_db), user=Depends(require_operator_or_admin)) -> CheckResponse:
+    try:
+        result = RoaPlannerService(RipeStatClient(db)).check(payload.prefix, payload.origin_as, payload.max_length)
+        response = _store_and_respond(db, "roa-preflight", payload.prefix, payload.origin_as, result, user.id, payload.change_case_id)
+        write_audit_log(db, user_id=user.id, action='roa_preflight_checked', target_type='check', target_id=str(response.report_id), details_json={'prefix': payload.prefix, 'origin_as': payload.origin_as, 'max_length': payload.max_length, 'change_case_id': payload.change_case_id})
+        return response
+    except HTTPException:
+        raise
+    except Exception as exc:
+        raise HTTPException(status_code=500, detail=f"ROA-Preflight-Prüfung fehlgeschlagen: {exc}") from exc
+
+
 @router.post('/preflight', response_model=CheckResponse)
 def check_preflight(payload: PreflightCheckRequest, db: Session = Depends(get_db), user=Depends(require_operator_or_admin)) -> CheckResponse:
     try:

backend/app/api/routes_system.py

@@ -12,7 +12,7 @@
 def system_info():
     return {
         'name': 'RouteForge',
-        'version': 'v0.7.1-beta',
+        'version': 'v0.7.2-beta',
         'demo_mode': settings.demo_mode,
         'read_only': True,
         'data_sources': ['RIPEstat', 'RIPEstat Whois/Registry'],

backend/app/schemas.py

@@ -91,6 +91,25 @@ def valid_planned_origin(cls, v: str) -> str:
 
 
 
+
+class RoaPreflightCheckRequest(BaseModel):
+    change_case_id: int | None = None
+    prefix: str
+    origin_as: str
+    max_length: int | None = None
+
+    @field_validator("prefix")
+    @classmethod
+    def valid_prefix(cls, v: str) -> str:
+        validate_prefix(v)
+        return v
+
+    @field_validator("origin_as")
+    @classmethod
+    def valid_origin(cls, v: str) -> str:
+        normalize_asn(v)
+        return v
+
 class BgpVisibilityCheckRequest(BaseModel):
     change_case_id: int | None = None
     prefix: str

backend/app/services/roa_planner_service.py

@@ -0,0 +1,136 @@
+import ipaddress
+
+from app.core.normalize import format_asn, normalize_asn, validate_prefix
+from app.services.ripe_stat_client import RipeStatClient
+
+
+class RoaPlannerService:
+    def __init__(self, client: RipeStatClient):
+        self.client = client
+
+    def check(self, prefix: str, origin_as: str, max_length: int | None = None) -> dict:
+        normalized_prefix = validate_prefix(prefix)
+        normalized_origin = format_asn(normalize_asn(origin_as))
+        network = ipaddress.ip_network(normalized_prefix, strict=False)
+        prefix_length = network.prefixlen
+        effective_max_length = max_length if max_length is not None else prefix_length
+
+        diagnostics: list[dict] = []
+        recommendations: list[str] = []
+        matching_roas: list[dict] = []
+        conflicting_roas: list[dict] = []
+
+        if effective_max_length < prefix_length:
+            return {
+                "status": "CRITICAL",
+                "summary": "Max Length ist kleiner als die Prefixlänge und damit ungültig.",
+                "recommendations": ["Setzen Sie max_length mindestens auf die Prefixlänge.", "Prüfen Sie Prefix und Eingaben auf Tippfehler."],
+                "details": {
+                    "prefix": normalized_prefix,
+                    "origin_as": normalized_origin,
+                    "max_length": max_length,
+                    "effective_max_length": effective_max_length,
+                    "prefix_length": prefix_length,
+                    "planned_validation_state": "invalid",
+                    "matching_roas": matching_roas,
+                    "conflicting_roas": conflicting_roas,
+                    "suggested_roa": {"prefix": normalized_prefix, "origin_as": normalized_origin, "max_length": prefix_length},
+                    "max_length_risk": "invalid_too_small",
+                    "recommendations": recommendations,
+                    "source_diagnostics": diagnostics,
+                },
+            }
+
+        roas_payload, roas_diag = self.client.get_with_diagnostics("rpki-validation", {"resource": normalized_prefix})
+        diagnostics.append(roas_diag)
+        if not roas_payload or not isinstance(roas_payload, dict):
+            return self._unknown(normalized_prefix, normalized_origin, max_length, effective_max_length, prefix_length, diagnostics)
+
+        validating_roas = (((roas_payload.get("data") or {}).get("validating_roas")) or [])
+        if not isinstance(validating_roas, list):
+            validating_roas = []
+
+        for roa in validating_roas:
+            try:
+                roa_origin = format_asn(normalize_asn(str(roa.get("origin") or roa.get("origin_asn") or "")))
+                roa_prefix = str(roa.get("prefix") or "")
+                roa_max_len = int(roa.get("max_length") or roa.get("maxlength") or ipaddress.ip_network(roa_prefix, strict=False).prefixlen)
+            except Exception:
+                continue
+            record = {"prefix": roa_prefix, "origin_as": roa_origin, "max_length": roa_max_len}
+            if roa_origin == normalized_origin and roa_max_len >= prefix_length:
+                matching_roas.append(record)
+            elif roa_origin != normalized_origin:
+                conflicting_roas.append(record)
+            elif roa_max_len < prefix_length:
+                conflicting_roas.append({**record, "reason": "max_length_too_small"})
+
+        planned_validation_state = "not_found"
+        status = "WARNING"
+        summary = "Keine passende ROA gefunden; ein read-only Vorschlag wurde erstellt."
+        max_length_risk = "none"
+
+        if matching_roas:
+            planned_validation_state = "valid"
+            status = "OK"
+            summary = "Geplantes Announcement ist durch mindestens eine passende ROA abgedeckt."
+            recommendations.append("Bestehende ROA-Abdeckung ist vorhanden; Änderungen außerhalb von RouteForge nur bei Bedarf durchführen.")
+
+        if conflicting_roas and not matching_roas:
+            planned_validation_state = "invalid"
+            status = "CRITICAL"
+            summary = "Konfliktierende ROAs deuten auf ein invalides geplantes Announcement hin."
+            recommendations.append("Origin-AS oder Prefixplanung prüfen; bestehende ROA-Konflikte außerhalb von RouteForge bereinigen.")
+
+        if effective_max_length > prefix_length + 2:
+            max_length_risk = "broad"
+            if status == "OK":
+                status = "WARNING"
+            recommendations.append("Max Length ist relativ breit gewählt; reduzieren Sie die Breite, um Hijack-Risiko zu senken.")
+
+        suggested_roa = None if matching_roas else {"prefix": normalized_prefix, "origin_as": normalized_origin, "max_length": effective_max_length}
+        if suggested_roa:
+            recommendations.append("ROA-Vorschlag extern durch LIR/Operator prüfen und außerhalb von RouteForge umsetzen.")
+
+        return {
+            "status": status,
+            "summary": summary,
+            "recommendations": recommendations or ["Keine zusätzlichen Empfehlungen."],
+            "details": {
+                "prefix": normalized_prefix,
+                "origin_as": normalized_origin,
+                "max_length": max_length,
+                "effective_max_length": effective_max_length,
+                "prefix_length": prefix_length,
+                "planned_validation_state": planned_validation_state,
+                "matching_roas": matching_roas,
+                "conflicting_roas": conflicting_roas,
+                "suggested_roa": suggested_roa,
+                "max_length_risk": max_length_risk,
+                "recommendations": recommendations,
+                "source_diagnostics": diagnostics,
+            },
+            "input": {"prefix": normalized_prefix, "origin_as": normalized_origin, "max_length": max_length},
+        }
+
+    def _unknown(self, prefix: str, origin_as: str, max_length: int | None, effective_max_length: int, prefix_length: int, diagnostics: list[dict]) -> dict:
+        return {
+            "status": "UNKNOWN",
+            "summary": "RPKI-Datenquelle liefert aktuell keine belastbare Aussage.",
+            "recommendations": ["Später erneut prüfen.", "Externe ROA-Quelle manuell validieren."],
+            "details": {
+                "prefix": prefix,
+                "origin_as": origin_as,
+                "max_length": max_length,
+                "effective_max_length": effective_max_length,
+                "prefix_length": prefix_length,
+                "planned_validation_state": "unknown",
+                "matching_roas": [],
+                "conflicting_roas": [],
+                "suggested_roa": {"prefix": prefix, "origin_as": origin_as, "max_length": effective_max_length},
+                "max_length_risk": "unknown",
+                "recommendations": ["Später erneut prüfen.", "Externe ROA-Quelle manuell validieren."],
+                "source_diagnostics": diagnostics,
+            },
+            "input": {"prefix": prefix, "origin_as": origin_as, "max_length": max_length},
+        }

backend/tests/test_api_smoke.py

@@ -502,3 +502,64 @@ def get_with_diagnostics(self, endpoint, _): return self.responses.get(endpoint,
     assert svc.check('198.51.100.0/24', 'AS3320')['status'] == 'CRITICAL'
     svc2 = BgpVisibilityService(C({'routing-status': ({'data': {'routes': [{'origin': 'AS3320'}, {'origin': 'AS64496'}]}}, {}), 'bgp-state': ({'data': {}}, {})}))
     assert svc2.check('198.51.100.0/24', None)['status'] == 'WARNING'
+
+
+def test_roa_preflight_roles_and_behavior() -> None:
+    client = _client()
+    _setup_and_login(client)
+    assert client.post('/api/check/roa-preflight', json={'prefix':'192.0.2.0/24','origin_as':'AS3320','max_length':24}).status_code == 200
+
+    create_op = client.post('/api/users', json={'username': 'op_roa', 'email': 'op_roa@example.org', 'password': 'OperatorPass123!', 'role': 'operator'})
+    assert create_op.status_code == 200
+    create_vw = client.post('/api/users', json={'username': 'vw_roa', 'email': 'vw_roa@example.org', 'password': 'ViewerPass123!', 'role': 'viewer'})
+    assert create_vw.status_code == 200
+
+    client.post('/api/auth/logout')
+    assert client.post('/api/auth/login', json={'username':'op_roa','password':'OperatorPass123!'}).status_code == 200
+    assert client.post('/api/check/roa-preflight', json={'prefix':'192.0.2.0/24','origin_as':'AS3320'}).status_code == 200
+
+    client.post('/api/auth/logout')
+    assert client.post('/api/auth/login', json={'username':'vw_roa','password':'ViewerPass123!'}).status_code == 200
+    assert client.post('/api/check/roa-preflight', json={'prefix':'192.0.2.0/24','origin_as':'AS3320'}).status_code == 403
+
+
+def test_roa_preflight_change_case_and_audit() -> None:
+    client = _client()
+    _setup_and_login(client)
+    cc = client.post('/api/change-cases', json={'title':'ROA case','description':'x'})
+    assert cc.status_code == 200
+    cc_id = cc.json()['id']
+
+    not_found = client.post('/api/check/roa-preflight', json={'prefix':'198.51.100.0/24','origin_as':'AS64496','change_case_id':999999})
+    assert not_found.status_code == 404
+
+    resp = client.post('/api/check/roa-preflight', json={'prefix':'198.51.100.0/24','origin_as':'AS64496','max_length':16,'change_case_id':cc_id})
+    assert resp.status_code == 200
+    body = resp.json()
+    assert body['status'] in {'OK','WARNING','CRITICAL','UNKNOWN'}
+    assert body['details']['prefix'] == '198.51.100.0/24'
+    assert body['details']['origin_as'] == 'AS64496'
+    assert body['details']['effective_max_length'] == 16
+    assert body['details']['max_length_risk'] in {'none','broad','invalid_too_small','unknown'}
+
+    reports = client.get(f'/api/change-cases/{cc_id}/reports')
+    assert reports.status_code == 200
+    assert any(r.get('check_type') == 'roa-preflight' for r in reports.json())
+
+    audit = client.get('/api/audit-log?limit=200')
+    assert audit.status_code == 200
+    actions = [i.get('action') for i in audit.json().get('items', [])]
+    assert 'roa_preflight_checked' in actions
+
+
+def test_roa_preflight_suggested_roa_and_invalid_max_length() -> None:
+    client = _client()
+    _setup_and_login(client)
+    warning = client.post('/api/check/roa-preflight', json={'prefix':'203.0.113.0/24','origin_as':'AS3320'})
+    assert warning.status_code == 200
+    wd = warning.json()['details']
+    assert 'suggested_roa' in wd
+
+    critical = client.post('/api/check/roa-preflight', json={'prefix':'203.0.113.0/24','origin_as':'AS3320','max_length':23})
+    assert critical.status_code == 200
+    assert critical.json()['status'] == 'CRITICAL'