Skip to main content

Building a Secure IIoT Edge Gateway: IEC 62443's Seven FRs in Working Code

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:

STRIDEThreatControl
SpoofingImpersonating an operatorSigned, expiring auth tokens
TamperingModifying a command/telemetry value in transitSigned message envelope + audit hash chain
RepudiationDenying an issued commandEvery action — success or denial — is audited before the function returns
Information DisclosureField data leaking to the wrong zoneZone/conduit allow-list + mutual TLS
Denial of ServiceFlooding the gatewayPer-actor rate limiting
Elevation of PrivilegeOperator performing an engineer-only actionServer-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.

Vivek US
Author
Vivek US
14+ years architecting secure, resilient Industrial IoT platforms — embedded systems, edge computing, mesh networking, and cloud-native infrastructure.