Introduction#
In my IEC 62443 deep dive I walked through the standard’s seven Foundational Requirements (FRs) and how they map onto firmware decisions. That post was theory. This one is the other half: an actual Modbus-to-MQTT edge gateway where every one of those seven FRs is implemented as real, tested code — not a checklist item, a control that fails its own automated test if you weaken it.
GitHub Repository: github.com/iam-vivekus/iec62443-secure-iiot-gateway
The field layer is a simulated Modbus device rather than physical hardware, deliberately. The security logic — auth, RBAC, audit, signing, rate limiting, zone enforcement — doesn’t change one line whether the field zone behind it is a simulator or a real ESP32 talking RS-485. Keeping it in software means anyone can clone the repo and reproduce every result with pip install -r requirements.txt && pytest — no bench setup required.
Architecture: Zones, Conduits, and One Trusted Bridge#
flowchart LR
subgraph FIELD["Field Zone"]
M[Modbus field device
holding registers]
end
subgraph GW["Gateway Zone (trusted bridge)"]
EG[EdgeGateway]
RBAC[Auth + RBAC]
RL[Rate limiter]
AL[Audit log
hash chain]
end
subgraph ENT["Enterprise / Cloud Zone"]
BROKER[MQTT broker
mutual TLS]
end
M <-->|conduit: command / telemetry| EG
EG --> RBAC --> RL --> AL
EG -->|conduit: signed telemetry / ack| BROKER
The design constraint that matters most here: the field zone has no conduit directly to the enterprise zone. That’s not just a line on a diagram — gateway/zones.py enforces it at runtime, and a test proves it:
def test_field_cannot_reach_enterprise_directly():
enforcer = ZoneEnforcer.default_topology()
with pytest.raises(ZoneViolationError):
enforcer.check(Zone.FIELD, Zone.ENTERPRISE, "telemetry")
Every message that crosses a boundary goes through the gateway, where authentication, authorization, rate limiting, and audit logging are mandatory pipeline steps, not optional middleware someone could forget to wire in.
FR1/FR2 — Authentication and Use Control#
Instead of a long-lived shared secret, actors present an HMAC-signed, time-limited token. A stolen token is only useful until it expires, and the signature check is constant-time to avoid leaking timing information:
def verify_token(secret: bytes, token: str) -> TokenClaims:
body_b64, signature = token.split(".")
expected_signature = _sign(secret, body_b64.encode())
if not hmac.compare_digest(signature, expected_signature):
raise AuthenticationError("Token signature invalid — possible tampering")
...
if time.time() > claims.expires_at:
raise AuthenticationError(f"Token for '{claims.subject}' expired")
Authorization is a server-side role → permission matrix — the gateway never trusts a caller’s self-declared role beyond what the signature actually verifies:
ROLE_PERMISSIONS = {
Role.OPERATOR: frozenset({Permission.READ_REGISTER, Permission.WRITE_REGISTER}),
Role.ENGINEER: frozenset({Permission.READ_REGISTER, Permission.WRITE_REGISTER, Permission.RECONFIGURE}),
Role.AUDITOR: frozenset({Permission.VIEW_AUDIT_LOG}),
}
An auditor token trying to write a register gets rejected before the field device is ever touched — and the rejection itself is logged, because a denied command is a security-relevant event in its own right.
FR3/FR4 — Integrity and Confidentiality#
TLS protects data between two endpoints, but it says nothing about whether a payload was legitimately produced before it entered the tunnel, or whether an intermediate broker altered it. So every outbound message also gets an application-layer HMAC signature that survives relaying through an untrusted intermediary:
def sign_payload(secret: bytes, payload: dict) -> dict:
body = json.dumps(payload, sort_keys=True).encode()
signature = hmac.new(secret, body, hashlib.sha256).hexdigest()
return {"payload": payload, "signature": signature, "signed_at": time.time()}
For the confidentiality side, the production transport (MqttTlsTransport) requires mutual TLS — both the gateway and the broker present certificates signed by the same local CA, generated with scripts/generate_demo_certs.sh. It’s not wired into the automated test suite on purpose: CI shouldn’t depend on a live broker being reachable, so tests exercise the identical signing/verification logic through an InMemoryTransport instead.
FR5 — Restricted Data Flow#
Covered above as the zone/conduit model, but worth calling out: the enforcement is a plain allow-list, not a clever heuristic —
def check(self, source: Zone, destination: Zone, message_type: str) -> None:
rule = self._rules.get((source, destination))
if rule is None:
raise ZoneViolationError(f"No conduit permits {source.value} -> {destination.value}")
if message_type not in rule.allowed_message_types:
raise ZoneViolationError(f"Conduit ... does not allow message type '{message_type}'")
Simple enough to audit by reading it, which is exactly the property you want from a control that’s supposed to contain blast radius.
FR6 — Tamper-Evident Audit Logging#
Every audit entry’s hash incorporates the previous entry’s hash — a hash chain. Rewriting any historical entry breaks every hash after it:
def verify_integrity(self) -> bool:
expected_prev = GENESIS_HASH
for entry in self._entries:
if entry.prev_hash != expected_prev or entry.compute_hash() != entry.entry_hash:
return False
expected_prev = entry.entry_hash
return True
The test that actually proves this matters more than the code:
def test_tampering_with_a_historical_entry_breaks_the_chain():
log = AuditLog()
log.append("alice", "read_register", "0", "success value=68")
log.append("alice", "write_register", "1", "success value=55")
assert log.verify_integrity() is True
log.tamper_for_demo(index=1, new_result="success value=99") # attacker rewrites history
assert log.verify_integrity() is False
An audit log you can silently edit isn’t an audit log — it’s a suggestion.
FR7 — Resource Availability#
A per-actor token-bucket rate limiter means one compromised or malfunctioning client can’t starve the field device or every other actor on the gateway:
def check(self, actor: str) -> None:
self._refill(actor)
if self._tokens[actor] < 1:
raise RateLimitExceededError(f"Rate limit exceeded for actor '{actor}'")
self._tokens[actor] -= 1
Budgets are tracked per actor specifically so a noisy operator doesn’t lock out an engineer who genuinely needs to push a config change.
Seeing It Run#
demo.py narrates all seven FRs firing in sequence against an in-memory transport — no broker, no hardware, just python demo.py:
=== 2. Auditor role is denied write access (FR2 — use control) ===
Rejected as expected: Role 'auditor' lacks permission 'write_register'
=== 5. Rate limiting stops a runaway client (FR7) ===
request 1: accepted
request 4: rejected — Rate limit exceeded for actor 'alice'
=== 7. Audit log is tamper-evident (FR2, FR6) ===
Chain intact before tampering: True
Chain intact after rewriting entry 0: False
And the full suite — 22 tests, one file per control area — passes in well under a second:
tests/test_audit_log.py ..... [ 13%]
tests/test_gateway.py ....... [ 50%]
tests/test_rate_limit.py ... [ 63%]
tests/test_rbac.py ..... [ 86%]
tests/test_zones.py ... [100%]
22 passed in 0.04s
Threat Model, in Brief#
The repo’s docs/threat-model.md runs a full STRIDE pass; the short version:
| STRIDE | Threat | Control |
|---|---|---|
| Spoofing | Impersonating an operator | Signed, expiring auth tokens |
| Tampering | Modifying a command/telemetry value in transit | Signed message envelope + audit hash chain |
| Repudiation | Denying an issued command | Every action — success or denial — is audited before the function returns |
| Information Disclosure | Field data leaking to the wrong zone | Zone/conduit allow-list + mutual TLS |
| Denial of Service | Flooding the gateway | Per-actor rate limiting |
| Elevation of Privilege | Operator performing an engineer-only action | Server-side permission check, never a trusted client claim |
Equally important is what the threat model documents as out of scope rather than pretending is solved: secrets rotation for the HMAC keys, physical security of the field device, and dependency supply-chain verification would all need dedicated tooling in a real deployment. Writing that down explicitly is part of what makes a threat model useful instead of decorative.
Conclusion#
Anyone can describe IEC 62443’s seven Foundational Requirements. What’s harder — and more useful when you’re actually building something — is having each one anchored to code that breaks its own tests the moment the control is weakened. That’s the point of this project: not a diagram, a working gateway you can clone, run, and try to break.
Key takeaways:
- FR1/FR2 — expiring signed tokens plus a server-side permission matrix, never a client-declared role
- FR3/FR4 — application-layer message signing as defense-in-depth on top of (not instead of) mutual TLS
- FR5 — zone/conduit enforcement as a plain allow-list simple enough to audit by reading it
- FR6 — a hash-chained audit log where tampering is detectable, not just logged
- FR7 — per-actor rate limiting so one bad client can’t starve everyone else
Full source, tests, architecture doc, and threat model: github.com/iam-vivekus/iec62443-secure-iiot-gateway.
