Setting Up Secure Data Boundaries for Academic IT: FAIR-Compliant Architecture & Automation #

import logging
from datetime import datetime, timezone
from typing import Any, Dict, Optional
from jsonschema import validate, ValidationError, SchemaError

# Configure structured logging for audit trails
logging.basicConfig(
    level=logging.INFO,
    format='{"timestamp":"%(asctime)s","level":"%(levelname)s","module":"%(module)s","message":"%(message)s"}'
)
logger = logging.getLogger("boundary_validator")

INGESTION_SCHEMA: Dict[str, Any] = {
    "$schema": "https://json-schema.org/draft/2020-12/schema",
    "type": "object",
    "required": ["access_level", "jurisdiction", "embargo_date", "pii_flag", "dataset_id"],
    "properties": {
        "access_level": {"type": "string", "enum": ["public", "restricted", "confidential", "embargoed"]},
        "jurisdiction": {"type": "string", "pattern": "^[A-Z]{2}$"},
        "embargo_date": {"type": "string", "format": "date"},
        "pii_flag": {"type": "boolean"},
        "dataset_id": {"type": "string", "pattern": "^DS-[0-9]{8}$"}
    },
    "additionalProperties": False
}

class BoundaryValidationError(Exception):
    """Custom exception for ingestion boundary violations."""
    def __init__(self, message: str, http_status: int = 400, details: Optional[Dict] = None):
        super().__init__(message)
        self.http_status = http_status
        self.details = details or {}

def validate_ingestion_metadata(payload: Dict[str, Any]) -> Dict[str, Any]:
    """
    Validates dataset metadata against institutional boundary schema.
    Returns sanitized payload or raises BoundaryValidationError.
    """
    try:
        validate(instance=payload, schema=INGESTION_SCHEMA)
    except SchemaError as e:
        logger.error("Schema configuration error: %s", str(e))
        raise BoundaryValidationError("Internal schema misconfiguration", http_status=500)
    except ValidationError as e:
        logger.warning("Metadata validation failed for payload: %s", e.message)
        raise BoundaryValidationError(
            message="Invalid boundary metadata",
            http_status=400,
            details={"field": e.json_path, "reason": e.message}
        )

    # Enforce logical boundary constraints beyond JSON structure
    if payload["access_level"] == "public" and payload["pii_flag"] is True:
        raise BoundaryValidationError(
            "Public datasets cannot contain PII",
            http_status=422,
            details={"conflict": "access_level vs pii_flag"}
        )
    
    embargo = datetime.fromisoformat(payload["embargo_date"]).replace(tzinfo=timezone.utc)
    if embargo < datetime.now(timezone.utc) and payload["access_level"] == "embargoed":
        raise BoundaryValidationError(
            "Embargo date has expired; update access_level",
            http_status=422,
            details={"expired_date": payload["embargo_date"]}
        )

    logger.info("Boundary validation passed for dataset %s", payload["dataset_id"])
    return payload

from dataclasses import dataclass
from typing import List, Dict, Any
from datetime import datetime, timezone

@dataclass(frozen=True)
class AccessRequest:
    user_id: str
    user_affiliation: str
    token_scopes: List[str]
    source_ip: str
    dataset_metadata: Dict[str, Any]

class PolicyDecisionPoint:
    """Evaluates access requests against FAIR-compliant boundary policies."""
    
    ALLOWED_AFFILIATIONS = {"university.edu", "research-institute.org"}
    RESTRICTED_SCOPES = {"data.read", "data.write", "data.admin"}

    def evaluate(self, request: AccessRequest) -> Dict[str, Any]:
        """Returns allow/deny decision with justification."""
        metadata = request.dataset_metadata

        # 1. Check PII boundary
        if metadata.get("pii_flag") and "pii.access" not in request.token_scopes:
            return {"decision": "deny", "code": 403, "reason": "Missing PII access scope"}

        # 2. Check jurisdictional compliance
        if metadata.get("jurisdiction") == "EU" and not self._is_gdpr_compliant(request):
            return {"decision": "deny", "code": 403, "reason": "GDPR jurisdiction mismatch"}

        # 3. Check embargo status
        if metadata.get("access_level") == "embargoed":
            embargo = datetime.fromisoformat(metadata["embargo_date"]).replace(tzinfo=timezone.utc)
            if datetime.now(timezone.utc) < embargo:
                return {"decision": "deny", "code": 403, "reason": "Dataset under active embargo"}

        # 4. Check affiliation boundary
        if metadata.get("access_level") in ("restricted", "confidential"):
            if not any(aff in request.user_affiliation for aff in self.ALLOWED_AFFILIATIONS):
                return {"decision": "deny", "code": 403, "reason": "Institutional affiliation required"}

        return {"decision": "allow", "code": 200, "reason": "Boundary constraints satisfied"}

    def _is_gdpr_compliant(self, request: AccessRequest) -> bool:
        """Placeholder for GDPR consent verification logic."""
        return "consent.verified" in request.token_scopes

from typing import Any, Dict
from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import JSONResponse
import time

app = FastAPI(title="Academic Data Boundary Router")
RATE_LIMIT_WINDOW = 60  # seconds
RATE_LIMIT_MAX = 100
request_log: Dict[str, list] = {}

@app.middleware("http")
async def enforce_rate_limit(request: Request, call_next):
    client_ip = request.client.host
    now = time.time()
    
    if client_ip not in request_log:
        request_log[client_ip] = []
    
    # Prune expired entries
    request_log[client_ip] = [t for t in request_log[client_ip] if now - t < RATE_LIMIT_WINDOW]
    
    if len(request_log[client_ip]) >= RATE_LIMIT_MAX:
        return JSONResponse(
            status_code=429,
            content={"detail": "Rate limit exceeded", "retry_after": RATE_LIMIT_WINDOW}
        )
    
    request_log[client_ip].append(now)
    response = await call_next(request)
    response.headers["X-RateLimit-Remaining"] = str(RATE_LIMIT_MAX - len(request_log[client_ip]))
    return response

@app.post("/api/v1/datasets/access")
async def route_access_request(payload: Dict[str, Any]):
    # In production, extract JWT claims, validate signatures, and instantiate AccessRequest
    # This example demonstrates routing logic post-PDP evaluation.
    try:
        # Simulate PDP evaluation
        decision = {"decision": "allow", "code": 200}
        
        if decision["decision"] == "allow":
            return {"status": "authorized", "data_uri": "s3://secure-bucket/ds-12345678"}
        elif decision["decision"] == "deny":
            raise HTTPException(status_code=403, detail=decision["reason"])
        else:
            # Fallback: route to manual compliance review
            return JSONResponse(status_code=202, content={"status": "queued_for_review", "ticket_id": "REV-9921"})
    except HTTPException:
        # Preserve intentional policy responses (e.g., 403 denials) instead of masking them as 500.
        raise
    except Exception:
        raise HTTPException(status_code=500, detail="Boundary routing failure")