Add Azure Data Link Layer assurance and MACsec checks

CHANGELOG.md

@@ -9,6 +9,7 @@ OpenShield uses [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 ### Added
 
+- Azure Data Link Layer Assurance API with LLC and MAC coverage plus ExpressRoute Direct MACsec checks
 - Azure public-cloud Physical Layer Assurance API with complete OSI and IEEE PHY domain, sublayer, and provider-evidence coverage
 - Semgrep SAST integrated into GitHub Actions CI as an open-source, account-free complement to CodeQL
 - OpenSSF Best Practices Passing Badge achieved with 100% of applicable Passing-level criteria completed

api/routes/assurance.py

@@ -5,6 +5,7 @@
 from flask import Blueprint, jsonify
 
 from api.services.physical_assurance import CatalogValidationError, get_physical_assurance_report
+from api.services.data_link_assurance import get_data_link_assurance_report
 
 
 assurance_bp = Blueprint("assurance", __name__)
@@ -22,3 +23,16 @@ def get_physical_layer_assurance():
     except Exception as exc:
         logger.error("Failed to build physical assurance report: %s", exc)
         return jsonify({"error": "Physical assurance report generation failed"}), 500
+
+
+@assurance_bp.get("/api/assurance/data-link-layer")
+def get_data_link_layer_assurance():
+    """Return Azure public-cloud OSI Layer 2 responsibility and evidence coverage."""
+    try:
+        return jsonify(get_data_link_assurance_report())
+    except CatalogValidationError as exc:
+        logger.error("Data Link assurance catalog validation failed: %s", exc)
+        return jsonify({"error": "Data Link assurance catalog is unavailable"}), 500
+    except Exception as exc:
+        logger.error("Failed to build Data Link assurance report: %s", exc)
+        return jsonify({"error": "Data Link assurance report generation failed"}), 500

api/services/assurance_catalog.py

@@ -0,0 +1,73 @@
+"""Shared validation primitives for closed assurance catalogs."""
+
+from __future__ import annotations
+
+from datetime import date
+from typing import Any
+from urllib.parse import urlparse
+
+
+class CatalogValidationError(ValueError):
+    """Raised when a bundled assurance catalog is incomplete or unsafe."""
+
+
+def require_string(item: dict[str, Any], field: str, context: str) -> str:
+    """Return a required non-empty string field."""
+    value = item.get(field)
+    if not isinstance(value, str) or not value.strip():
+        raise CatalogValidationError(f"{context}: {field} must be a non-empty string")
+    return value
+
+
+def require_unique(items: Any, name: str) -> dict[str, dict[str, Any]]:
+    """Index a required object list by unique string ID."""
+    if not isinstance(items, list):
+        raise CatalogValidationError(f"{name} must be a list")
+    indexed: dict[str, dict[str, Any]] = {}
+    for item in items:
+        if not isinstance(item, dict):
+            raise CatalogValidationError(f"{name}: every entry must be an object")
+        item_id = require_string(item, "id", name)
+        if item_id in indexed:
+            raise CatalogValidationError(f"{name}: duplicate id {item_id}")
+        indexed[item_id] = item
+    return indexed
+
+
+def require_reference_list(item: dict[str, Any], field: str, allowed_ids: set[str], context: str) -> list[str]:
+    """Validate a non-empty list of unique cross-references."""
+    values = item.get(field)
+    if not isinstance(values, list) or not values:
+        raise CatalogValidationError(f"{context}: {field} must be a non-empty list")
+    if any(not isinstance(value, str) for value in values):
+        raise CatalogValidationError(f"{context}: {field} must contain only strings")
+    if len(values) != len(set(values)):
+        raise CatalogValidationError(f"{context}: {field} contains duplicate references")
+    unknown = set(values) - allowed_ids
+    if unknown:
+        raise CatalogValidationError(f"{context}: {field} contains unknown ids {sorted(unknown)}")
+    return values
+
+
+def parse_iso_date(value: Any, context: str) -> date:
+    """Parse a required ISO-8601 calendar date."""
+    if not isinstance(value, str):
+        raise CatalogValidationError(f"{context}: date must be an ISO-8601 string")
+    try:
+        return date.fromisoformat(value)
+    except ValueError as exc:
+        raise CatalogValidationError(f"{context}: invalid ISO-8601 date {value!r}") from exc
+
+
+def validate_evidence_source(source: dict[str, Any], source_id: str, allowed_hosts: set[str]) -> tuple[date, date]:
+    """Validate common evidence metadata and return its review dates."""
+    require_string(source, "title", source_id)
+    url = require_string(source, "url", source_id)
+    parsed_url = urlparse(url)
+    if parsed_url.scheme != "https" or parsed_url.hostname not in allowed_hosts:
+        raise CatalogValidationError(f"{source_id}: evidence URL must use HTTPS on an allowed host")
+    reviewed_at = parse_iso_date(source.get("reviewed_at"), f"{source_id}.reviewed_at")
+    review_due_at = parse_iso_date(source.get("review_due_at"), f"{source_id}.review_due_at")
+    if review_due_at <= reviewed_at:
+        raise CatalogValidationError(f"{source_id}: review_due_at must be after reviewed_at")
+    return reviewed_at, review_due_at

api/services/data_link_assurance.py

@@ -0,0 +1,153 @@
+"""Load, validate, and report Azure Data Link assurance coverage."""
+
+from __future__ import annotations
+
+import copy
+import json
+from datetime import date
+from pathlib import Path
+from typing import Any
+
+from api.services.assurance_catalog import (
+    CatalogValidationError,
+    require_reference_list,
+    require_string,
+    require_unique,
+    validate_evidence_source,
+)
+
+CATALOG_PATH = Path(__file__).resolve().parents[2] / "compliance" / "assurance" / "data_link_layer.json"
+EXPECTED_DOMAIN_IDS = {f"DL-{number:02d}" for number in range(1, 20)}
+EXPECTED_SUBLAYER_IDS = {"LLC", "MAC"}
+ALLOWED_RESPONSIBILITIES = {"Microsoft", "Customer", "Shared"}
+ALLOWED_APPLICABILITY = {"APPLICABLE", "NOT_APPLICABLE", "UNSUPPORTED"}
+ALLOWED_VERIFICATION = {
+    "PROVIDER_ATTESTED",
+    "PLATFORM_ENFORCED",
+    "AUTOMATICALLY_CHECKED",
+    "MANUALLY_VERIFIABLE",
+    "UNSUPPORTED",
+    "NOT_APPLICABLE",
+}
+
+
+def validate_catalog(catalog: dict[str, Any]) -> None:
+    """Fail closed when any Layer 2 domain, sublayer, decision, or cross-reference is missing."""
+    if not isinstance(catalog, dict):
+        raise CatalogValidationError("Catalog root must be an object")
+    layer = catalog.get("layer")
+    if not isinstance(layer, dict) or layer.get("number") != 2 or layer.get("name") != "Data Link":
+        raise CatalogValidationError("Catalog must describe OSI Layer 2 Data Link")
+    require_string(catalog, "catalog_version", "catalog")
+    scope = catalog.get("scope")
+    if not isinstance(scope, dict):
+        raise CatalogValidationError("scope must be an object")
+    require_string(scope, "statement", "scope")
+    require_string(scope, "responsibility_boundary", "scope")
+
+    domains = require_unique(catalog.get("domains"), "domains")
+    sublayers = require_unique(catalog.get("sublayers"), "sublayers")
+    evidence = require_unique(catalog.get("evidence_sources"), "evidence_sources")
+    automated = require_unique(catalog.get("automated_controls"), "automated_controls")
+    if set(domains) != EXPECTED_DOMAIN_IDS:
+        raise CatalogValidationError("domains must contain the complete DL-01 through DL-19 set")
+    if set(sublayers) != EXPECTED_SUBLAYER_IDS:
+        raise CatalogValidationError("sublayers must contain LLC and MAC")
+
+    for sublayer_id, sublayer in sublayers.items():
+        require_string(sublayer, "name", sublayer_id)
+        require_string(sublayer, "description", sublayer_id)
+    for evidence_id, source in evidence.items():
+        validate_evidence_source(source, evidence_id, {"learn.microsoft.com"})
+
+    referenced_sublayers: set[str] = set()
+    referenced_evidence: set[str] = set()
+    referenced_automated: set[str] = set()
+    for domain_id, domain in domains.items():
+        require_string(domain, "name", domain_id)
+        require_string(domain, "observability_method", domain_id)
+        responsibility = require_string(domain, "responsibility_owner", domain_id)
+        applicability = require_string(domain, "azure_applicability", domain_id)
+        verification = require_string(domain, "verification", domain_id)
+        if responsibility not in ALLOWED_RESPONSIBILITIES:
+            raise CatalogValidationError(f"{domain_id}: invalid responsibility_owner")
+        if applicability not in ALLOWED_APPLICABILITY:
+            raise CatalogValidationError(f"{domain_id}: invalid azure_applicability")
+        if verification not in ALLOWED_VERIFICATION:
+            raise CatalogValidationError(f"{domain_id}: invalid verification")
+        referenced_sublayers.update(require_reference_list(domain, "sublayer_ids", set(sublayers), domain_id))
+        referenced_evidence.update(require_reference_list(domain, "evidence_source_ids", set(evidence), domain_id))
+        control_ids = domain.get("automated_control_ids", [])
+        if verification == "AUTOMATICALLY_CHECKED":
+            referenced_automated.update(
+                require_reference_list(domain, "automated_control_ids", set(automated), domain_id)
+            )
+        elif control_ids:
+            raise CatalogValidationError(f"{domain_id}: only automatically checked domains may reference rules")
+    if referenced_sublayers != set(sublayers):
+        raise CatalogValidationError("LLC and MAC must both be covered")
+    if referenced_evidence != set(evidence):
+        raise CatalogValidationError("every evidence source must be referenced")
+    if referenced_automated != set(automated):
+        raise CatalogValidationError("every automated control must be cross-referenced")
+
+    for control_id, control in automated.items():
+        require_string(control, "name", control_id)
+        require_string(control, "playbook", control_id)
+        require_reference_list(control, "domain_ids", set(domains), control_id)
+        frameworks = control.get("frameworks")
+        if not isinstance(frameworks, dict) or set(frameworks) != {"CIS", "NIST", "ISO27001", "SOC2"}:
+            raise CatalogValidationError(f"{control_id}: all required framework mappings must be present")
+
+
+def load_catalog(path: Path = CATALOG_PATH) -> dict[str, Any]:
+    """Load and validate the bundled catalog."""
+    try:
+        with path.open(encoding="utf-8") as catalog_file:
+            catalog = json.load(catalog_file)
+    except (OSError, json.JSONDecodeError) as exc:
+        raise CatalogValidationError(f"Unable to load Data Link assurance catalog: {exc}") from exc
+    validate_catalog(catalog)
+    return catalog
+
+
+def build_report(catalog: dict[str, Any], as_of: date | None = None) -> dict[str, Any]:
+    """Build static responsibility coverage with independent evidence freshness."""
+    validate_catalog(catalog)
+    report = copy.deepcopy(catalog)
+    as_of = as_of or date.today()
+    evidence = {item["id"]: item for item in report["evidence_sources"]}
+    current = sum(date.fromisoformat(item["review_due_at"]) >= as_of for item in evidence.values())
+    for domain in report["domains"]:
+        domain["evidence"] = [evidence[source_id] for source_id in domain["evidence_source_ids"]]
+    report["catalog_coverage"] = {
+        "domains_covered": len(report["domains"]),
+        "domains_total": 19,
+        "sublayers_covered": 2,
+        "sublayers_total": 2,
+        "percent": 100,
+    }
+    report["evidence_freshness"] = {
+        "current_sources": current,
+        "total_sources": len(evidence),
+        "percent": round(current / len(evidence) * 100) if evidence else 0,
+        "assessed_as_of": as_of.isoformat(),
+    }
+    report["provider_assurance_state"] = "DOCUMENTED"
+    report["platform_enforcement_state"] = "DOCUMENTED_NOT_LIVE_INSPECTED"
+    report["automated_control_applicability"] = {
+        "state": "REQUIRES_SUBSCRIPTION_INVENTORY",
+        "without_expressroute_direct": "NOT_APPLICABLE",
+        "api_or_permission_failure": "INDETERMINATE",
+    }
+    report["limitations"] = [
+        "This report is not a live inspection of Microsoft switches, forwarding tables, VLANs, or fabric internals.",
+        "Provider-owned assurance domains do not create findings or change the tenant security score.",
+        "Only ExpressRoute Direct management-plane configuration is automatically checked.",
+    ]
+    return report
+
+
+def get_data_link_assurance_report(as_of: date | None = None) -> dict[str, Any]:
+    """Return the bundled Data Link assurance report."""
+    return build_report(load_catalog(), as_of=as_of)

api/services/physical_assurance.py

@@ -7,7 +7,14 @@
 from datetime import date
 from pathlib import Path
 from typing import Any
-from urllib.parse import urlparse
+
+from api.services.assurance_catalog import (
+    CatalogValidationError,
+    require_reference_list as _require_reference_list,
+    require_string as _require_string,
+    require_unique as _require_unique,
+    validate_evidence_source,
+)
 
 
 CATALOG_PATH = Path(__file__).resolve().parents[2] / "compliance" / "assurance" / "physical_layer.json"
@@ -32,52 +39,6 @@
 ALLOWED_EVIDENCE_HOSTS = {"learn.microsoft.com"}
 
 
-class CatalogValidationError(ValueError):
-    """Raised when the bundled assurance catalog is incomplete or unsafe."""
-
-
-def _require_string(item: dict[str, Any], field: str, context: str) -> str:
-    value = item.get(field)
-    if not isinstance(value, str) or not value.strip():
-        raise CatalogValidationError(f"{context}: {field} must be a non-empty string")
-    return value
-
-
-def _require_unique(items: list[dict[str, Any]], name: str) -> dict[str, dict[str, Any]]:
-    indexed: dict[str, dict[str, Any]] = {}
-    for item in items:
-        if not isinstance(item, dict):
-            raise CatalogValidationError(f"{name}: every entry must be an object")
-        item_id = _require_string(item, "id", name)
-        if item_id in indexed:
-            raise CatalogValidationError(f"{name}: duplicate id {item_id}")
-        indexed[item_id] = item
-    return indexed
-
-
-def _require_reference_list(item: dict[str, Any], field: str, allowed_ids: set[str], context: str) -> list[str]:
-    values = item.get(field)
-    if not isinstance(values, list) or not values:
-        raise CatalogValidationError(f"{context}: {field} must be a non-empty list")
-    if any(not isinstance(value, str) for value in values):
-        raise CatalogValidationError(f"{context}: {field} must contain only strings")
-    if len(values) != len(set(values)):
-        raise CatalogValidationError(f"{context}: {field} contains duplicate references")
-    unknown = set(values) - allowed_ids
-    if unknown:
-        raise CatalogValidationError(f"{context}: {field} contains unknown ids {sorted(unknown)}")
-    return values
-
-
-def _parse_date(value: Any, context: str) -> date:
-    if not isinstance(value, str):
-        raise CatalogValidationError(f"{context}: date must be an ISO-8601 string")
-    try:
-        return date.fromisoformat(value)
-    except ValueError as exc:
-        raise CatalogValidationError(f"{context}: invalid ISO-8601 date {value!r}") from exc
-
-
 def load_catalog(path: Path = CATALOG_PATH) -> dict[str, Any]:
     """Load and validate a physical assurance catalog from disk."""
     try:
@@ -153,16 +114,8 @@ def validate_catalog(catalog: dict[str, Any]) -> None:
         _require_reference_list(sublayer, "domain_ids", domain_ids, sublayer_id)
 
     for evidence_id, evidence in evidence_index.items():
-        _require_string(evidence, "title", evidence_id)
         _require_string(evidence, "evidence_type", evidence_id)
-        url = _require_string(evidence, "url", evidence_id)
-        parsed_url = urlparse(url)
-        if parsed_url.scheme != "https" or parsed_url.hostname not in ALLOWED_EVIDENCE_HOSTS:
-            raise CatalogValidationError(f"{evidence_id}: evidence URL must use HTTPS on an allowed host")
-        reviewed_at = _parse_date(evidence.get("reviewed_at"), f"{evidence_id}.reviewed_at")
-        review_due_at = _parse_date(evidence.get("review_due_at"), f"{evidence_id}.review_due_at")
-        if review_due_at <= reviewed_at:
-            raise CatalogValidationError(f"{evidence_id}: review_due_at must be after reviewed_at")
+        validate_evidence_source(evidence, evidence_id, ALLOWED_EVIDENCE_HOSTS)
 
     microsoft_controls: set[str] = set()
     iso_controls: set[str] = set()