Maritime Security Boundary Setup #

Maritime Security Boundary Setup is the discipline of turning International Ship and Port Facility Security (ISPS) Code mandates into executable, data-driven enforcement — modeling every geofence, restricted berth, and logical terminal as a typed policy object that gates vessel access, cargo movement, and audit routing in real time. Within the Core Maritime Architecture & Taxonomy framework, a security boundary is not a map overlay; it is a versioned contract that every crossing event must satisfy before downstream automation acts on it. Shipping operators and port authorities cannot tolerate silent bypasses: an ungated MARSEC-2 berth, a cargo type admitted without clearance, or an unlogged boundary crossing turns directly into a port state control finding. This page shows how to define zones as immutable Python structures, validate crossing events at the ingestion edge, quarantine ambiguous traffic, and keep the control plane available under load.

Ingestion Boundary & Protocol Handling #

Boundary-crossing events reach the enforcement layer from several transports, and each arrives with its own reliability and encoding profile. Position reports stream in from the AIS Data Stream Integration feed as NMEA-derived JSON; terminal operating system (TOS) webhooks push gate-in and berth-assignment events over HTTPS; optical character recognition at the gate emits container and vessel identifiers over an internal message bus. The security boundary must accept all of them at the wire, normalize them to a single event contract, and reject anything malformed before a single access decision is computed — the same syntax-tolerant, semantics-strict posture used across the wider architecture.

Each transport carries its own boundary rules:

• AIS feed — high-volume, lossy, and occasionally drifts past valid WGS-84 ranges (latitude beyond ±90, longitude beyond ±180). Clamp coordinates and reject reports whose MMSI cannot be resolved to an IMO number before evaluating a geofence.
• TOS webhooks — authenticated but bursty; deduplicate on the carrier-supplied event id because retransmission during broker redelivery is routine, not exceptional.
• Gate OCR / bus events — trustworthy for identity but weak on timing; treat the ingestion timestamp, not the payload timestamp, as authoritative for dwell-time enforcement.

The ingestion layer therefore performs four deterministic actions before any zone logic runs: authenticate the transport and record the source system, coerce timestamps to timezone-aware UTC, clamp or reject out-of-range coordinates, and emit a structured log line carrying correlation_id, source_system, and zone_id. Only a structurally sound, fully typed event is handed to the evaluation engine; everything else is dead-lettered with a SCHEMA_VIOLATION status so no partial decision ever propagates.

Python Data Structure Mapping #

Geospatial and logical security zones must be modeled as immutable typed structures that map directly to terminal infrastructure and ISPS clearance tiers. Loose dictionaries are unacceptable at this boundary: deploy Pydantic models that enforce coordinate precision, polygon closure, and access-control lists (ACLs) at instantiation, so a malformed zone definition fails at load time rather than during a live vessel approach. All code uses type annotations and structlog for structured JSON logging.

from __future__ import annotations

from enum import Enum
from typing import Literal

import structlog
from pydantic import BaseModel, Field, field_validator

log = structlog.get_logger()


class MARSECLevel(str, Enum):
    """Maritime security (MARSEC) levels — ISPS Code Part A."""
    LEVEL_1 = "1"  # normal operating posture
    LEVEL_2 = "2"  # heightened threat, additional protective measures
    LEVEL_3 = "3"  # probable or imminent security incident


class SecurityZone(BaseModel):
    zone_id: str = Field(..., pattern=r"^Z-\d{4}$")
    boundary_type: Literal["geofence", "logical_terminal", "restricted_berth"]
    # A closed ring needs at least 4 points (first vertex repeated as last).
    coordinates: list[tuple[float, float]] = Field(..., min_length=4)
    clearance_level: MARSECLevel
    authorized_vessel_classes: list[str]
    authorized_cargo_types: list[str]
    max_dwell_minutes: int = Field(ge=0)

    @field_validator("coordinates")
    @classmethod
    def validate_polygon_closure(
        cls, v: list[tuple[float, float]]
    ) -> list[tuple[float, float]]:
        if v[0] != v[-1]:
            raise ValueError("Geofence polygon must be closed (first vertex == last)")
        log.debug("zone_polygon_validated", vertices=len(v))
        return v

The inbound crossing event is modeled as its own strict type. Constructing it is the structural validation gate: a missing field, a nine-digit MMSI in the IMO slot, or an unparseable MARSEC value raises a ValidationError and routes to the dead-letter queue before any policy check executes.

from datetime import datetime


class BoundaryCrossing(BaseModel):
    vessel_imo: int = Field(..., ge=1_000_000, le=9_999_999)  # 7-digit IMO
    zone_id: str = Field(..., pattern=r"^Z-\d{4}$")
    security_clearance: MARSECLevel
    bol_id: str = Field(..., min_length=1)
    correlation_id: str = Field(..., min_length=1)
    observed_at: datetime

Field coercion follows the same conventions used elsewhere in the taxonomy: IMO numbers are validated against the 7-digit range (and, in production, the mod-11 check digit), UN/LOCODEs referenced by a zone are upper-cased five-character codes, and dwell windows are integer minutes. When a crossing references a specific box, its parent-child position resolves against the Container Hierarchy Data Models Vessel → Bay → Tier → Row → Container topology, so equipment-level segregation rules stay consistent as units move between zones.

Validation, Quarantine & Compliance Auditing #

Enforcement applies validation in three tiers, and the tier a failure lands in determines where the event is routed. The distinction is load-bearing: unrecoverable corruption must never share a queue with a recoverable business exception, because the two are triaged by different teams on different timelines.

1. Structural validation — the BoundaryCrossing model construction. A ValidationError here is a format defect (broken schema, out-of-range identifier) and fails fast to the dead-letter queue with SCHEMA_VIOLATION.
2. Semantic validation — clearance intersection. The vessel’s security clearance must meet or exceed the zone’s MARSEC level. A mismatch is recoverable (credentials may simply be stale), so it routes to a quarantine topic, not the DLQ.
3. Regulatory validation — cargo authorization. The cargo type resolved from the bill of lading must appear in the zone’s allow-list. Normalizing ISPS status against commercial documentation reuses the patterns in Bill of Lading Schema Mapping, so restricted cargo cannot enter a restricted berth without an explicit hold.

def _cargo_authorized(bol_id: str, authorized_cargo_types: list[str]) -> bool:
    """Resolve the bill of lading's cargo type and confirm it is permitted in
    the zone. Production deployments query the B/L service; the allow-list
    intersection is shown here."""
    # Placeholder: integrate with Bill of Lading Schema Mapping in production.
    return bool(authorized_cargo_types)


def evaluate_boundary_crossing(
    event: BoundaryCrossing, zone: SecurityZone
) -> dict[str, str]:
    """Apply clearance (semantic) then cargo (regulatory) checks. Structural
    validation has already succeeded by virtue of `event` being constructed."""
    if event.security_clearance < zone.clearance_level:
        log.warning(
            "boundary_quarantined",
            zone_id=zone.zone_id,
            reason="CLEARANCE_MISMATCH",
            vessel_imo=event.vessel_imo,
            correlation_id=event.correlation_id,
        )
        return {"status": "QUARANTINED", "reason": "CLEARANCE_MISMATCH",
                "trace_id": event.correlation_id}

    if not _cargo_authorized(event.bol_id, zone.authorized_cargo_types):
        log.warning(
            "boundary_blocked",
            zone_id=zone.zone_id,
            reason="CARGO_NOT_PERMITTED",
            correlation_id=event.correlation_id,
        )
        return {"status": "BLOCKED", "reason": "CARGO_NOT_PERMITTED",
                "trace_id": event.correlation_id}

    log.info(
        "boundary_allowed",
        zone_id=zone.zone_id,
        vessel_imo=event.vessel_imo,
        correlation_id=event.correlation_id,
    )
    return {"status": "ALLOWED", "trace_id": event.correlation_id}

Every terminal state — ALLOWED, QUARANTINED, BLOCKED, REJECTED — emits an immutable audit record capturing correlation_id, vessel_imo, zone_id, rule_applied, and validation_result. That chain is what satisfies port state control inspections and lets an auditor reconstruct exactly why any vessel was admitted or held, months after the fact. Enum ordering makes the clearance comparison meaningful: comparing MARSECLevel members lexically works because levels 1–3 are monotonically increasing, but production code should compare on an explicit integer rank to avoid depending on string ordering as the scheme evolves.

Downstream Integration #

A boundary decision is only useful if it propagates deterministically. An ALLOWED verdict releases the crossing to the Port Call Workflow Design state machine, which advances the vessel’s milestone timeline; a BLOCKED or QUARANTINED verdict raises a hold that the same state machine must honor before any gate move is authorized. Because both the workflow engine and the security layer consume from the same event stream, every cross-boundary publish is keyed on a deterministic (vessel_imo, zone_id, event_hash) tuple so that AIS retransmissions and webhook retries collapse to a no-op rather than double-triggering a stowage change.

At the API tier, the routing-facing surface of this layer is documented in Implementing ISPS security zones in routing APIs, where zone data is served from in-memory R-tree caches synchronized from a central policy registry and graduated proximity thresholds (advisory, recalculate, block) translate a raw distance into a compliance action. Downstream of a decision, cargo systems consuming Bill of Lading Schema Mapping output adjust stowage and crane assignments in lockstep, and the physical movement of each admitted vessel is correlated back against the live position feed from AIS Data Stream Integration. The security boundary is the authenticating edge for all of these publishes — no consumer accepts a state transition that did not pass through it.

Fallback Chains & Uptime Guarantees #

High-throughput port environments cannot absorb a hard pipeline halt; a security layer that stops evaluating stops the terminal. Decouple boundary validation from core transactional flows using an asynchronous broker (Kafka, RabbitMQ, or SQS), and make every consumer idempotent with a deduplication key of vessel_imo + zone_id + observed_at so broker redelivery never double-processes a crossing. Layer three explicit fallback tiers on top:

1. Transient failures — when an external TOS or customs lookup times out, retry with exponential backoff and jitter (base_delay * 2 ** attempt + random_jitter), capped at three attempts. Wrap each downstream in a circuit breaker that opens after consecutive 5xx/timeout errors and serves cached policy instead of hammering a degraded service.
2. Persistent schema violations — route to a dead-letter queue with full context preserved. DLQ consumers run reconciliation and alert the security operations dashboard; these events are never silently dropped.
3. Graceful degradation — when the primary policy registry is unreachable, fall back to cached clearance states under a strict TTL, tag the response DEGRADED_MODE=true, and require secondary manual verification before releasing restricted cargo. Degradation never means permissiveness: an unknown clearance under degraded mode holds, it does not pass.

The DLQ and the quarantine topic remain strictly separate. Quarantine holds events that parsed cleanly but failed a clearance or cargo check — a human or a registry refresh can resolve them, and they reconcile automatically once the authoritative source recovers. The DLQ holds corruption that requires engineering. Structured observability spans both: emit OpenTelemetry-compatible logs with trace_id and span_id, track p50/p95/p99 evaluation latency, quarantine rate, and DLQ depth, and alert on sustained thresholds.

Step-by-step Implementation Guide #

The reference pipeline below stands up a working boundary evaluator, from typed zone definitions to a resilient async consumer. Each step is runnable in isolation and uses type annotations with structlog.

Step 1 — Load and validate zone definitions #

Load the policy registry into typed SecurityZone objects at startup. A malformed zone must fail here, not during a live approach.

def load_zones(raw_zones: list[dict]) -> dict[str, SecurityZone]:
    zones: dict[str, SecurityZone] = {}
    for raw in raw_zones:
        zone = SecurityZone(**raw)  # raises ValidationError on bad polygon/ACL
        zones[zone.zone_id] = zone
    log.info("zones_loaded", count=len(zones))
    return zones

Step 2 — Normalize an inbound crossing event #

Coerce the transport payload into the strict BoundaryCrossing contract. Construction is the structural gate; a ValidationError here is a dead-letter, not a decision.

from pydantic import ValidationError


def normalize_event(payload: dict) -> BoundaryCrossing | None:
    try:
        return BoundaryCrossing(**payload)
    except ValidationError as exc:
        log.error("schema_violation", errors=exc.error_count(),
                  correlation_id=payload.get("correlation_id"))
        return None  # caller routes to the dead-letter queue

Step 3 — Resolve the zone and evaluate the crossing #

Look up the referenced zone and apply the clearance then cargo checks from evaluate_boundary_crossing.

def decide(event: BoundaryCrossing, zones: dict[str, SecurityZone]) -> dict[str, str]:
    zone = zones.get(event.zone_id)
    if zone is None:
        log.warning("zone_unknown", zone_id=event.zone_id,
                    correlation_id=event.correlation_id)
        return {"status": "QUARANTINED", "reason": "ZONE_UNKNOWN",
                "trace_id": event.correlation_id}
    return evaluate_boundary_crossing(event, zone)

Step 4 — Emit an immutable audit record #

Every terminal verdict writes a structured, append-only audit line keyed on the correlation id.

def audit(event: BoundaryCrossing, verdict: dict[str, str]) -> None:
    log.info(
        "boundary_audit",
        correlation_id=event.correlation_id,
        vessel_imo=event.vessel_imo,
        zone_id=event.zone_id,
        validation_result=verdict["status"],
        reason=verdict.get("reason", "OK"),
    )

Step 5 — Wrap the evaluator in a resilient consumer #

Compose the steps behind idempotency and a fallback so a single bad event never halts the stream.

def handle(payload: dict, zones: dict[str, SecurityZone]) -> dict[str, str]:
    event = normalize_event(payload)
    if event is None:
        return {"status": "REJECTED", "reason": "SCHEMA_VIOLATION"}
    verdict = decide(event, zones)
    audit(event, verdict)
    return verdict

Wire handle to an idempotent async consumer (dedup key vessel_imo + zone_id + observed_at), send REJECTED payloads to the DLQ and QUARANTINED ones to the quarantine topic, and only forward ALLOWED verdicts to the port-call workflow.

Troubleshooting Common Failures #

Frequently Asked Questions #

Related #

• Implementing ISPS security zones in routing APIs — the routing-tier surface that serves zones from spatial indexes and applies graduated thresholds
• Bill of Lading Schema Mapping — the commercial documentation layer cargo authorization checks against
• Container Hierarchy Data Models — the equipment topology that keeps segregation rules consistent across zone transitions
• Port Call Workflow Design — the idempotent state machine that consumes ALLOWED verdicts and honors holds
• AIS Data Stream Integration — the position feed that supplies and corroborates boundary-crossing events

Up: Core Maritime Architecture & Taxonomy — the parent framework governing schema, state, and boundary controls.