Skip to content
J.R.
日本語 EN 中文

03 / 05 · 2022.08 — 2023.07

Zhejiang Mulian Technology Co., Ltd.

Python backend in industrial-control security: the traffic pipeline, allow-list detection engine and Django REST API of an out-of-band audit system, with Suricata integrated for deny-list detection and deep parsing of 27 industrial protocols.

Role
Python Developer
Industry
Industrial control security
Location
Zhejiang
Team
13 people
Tenure
1 yr

Key results

  • 27

    Industrial protocols parsed in depth: Modbus, S7COMM, IEC104, IEC61850, DNP3, OPC UA, PROFINET, BACnet and more

  • 996

    Industrial vulnerability signatures loaded into Suricata for deny-list detection

  • 10 s

    Flow-aggregation window built on Redis key-expiry events, turning per-packet writes into per-flow writes

Projects here

PROJECTS · 1

2022.08 – 2023.07 · Python backend developer · Team 13

Industrial-control security audit platform

An out-of-band traffic-audit system for power and manufacturing control networks: probes capture mirrored switch traffic, Kafka feeds a Cython protocol-parsing engine, and three layers of policy — assets, allow-list, deny-list — drive detection and alerting. It parses 27 industrial protocols in depth (Modbus, S7COMM, IEC104, IEC61850 MMS/GOOSE/SV, DNP3, OPC UA/DA/AE, PROFINET, ENIP/CIP, BACnet and more) and has been deployed at power-sector research institutes.

  • Worked on the traffic pipeline: capture probe → Kafka → protocol-parsing engine → PostgreSQL; used Redis key-expiry events for 10-second flow aggregation and batch inserts, sharply cutting database writes, and Redis TTLs for device-offline detection
  • Implemented the allow-list detection engine keyed on 5-tuple, MAC and protocol function code, with a bitmap that separates "behaviour anomalies" from "content anomalies"; deep rule matching on IEC104 ASDU and OPC interface/function codes, plus automatic allow-list learning over a chosen time window
  • Built the Django REST endpoints for asset management, allow/deny-list policies, alerts and reports; Celery Beat jobs for report export, policy backup, expired-data cleanup and probe health checks
  • Integrated Suricata, parsed unified2 alerts and loaded nearly a thousand (996) industrial-vulnerability signatures for deny-list detection
  • Synced data to a Pilosa bitmap index through Kafka Connect (Debezium CDC) to back traffic statistics and multi-dimensional queries
  • (Customised edition) Parsed IEC 61850 SCD files to generate substation IED communication baselines and topology automatically
  • Python
  • Django / DRF
  • Celery
  • Kafka
  • Redis
  • PostgreSQL
  • Pilosa
  • Suricata
  • Cython
  • Vue 2

System architecture

Capture · probe Message bus Parse · detect Store · sync Application · management Offline alerts HTTP REST rules · assets · protocol switches · thresholds — pulled every 5 s (hot reload) Industrial-network mirrored trafficSPAN port · read-only network-capturer Capture probe · Python Scapy L2 raw socket selector polling over multiple NICs Cython Cstat counters deque buffer → Producer Operations safeguards Supervisor keeps daemons alive Celery checks probes every 5 min Capture / Kafka-write dual counters systemd checks Pilosa status cpulimit caps process CPU Suricata 4.0.4 AF_PACKET · cluster_flow 996 industrial vulnerability signatures Writes unified2 binary logs Kafka cluster cp-kafka 5.4 · ZooKeeper SASL auth · Docker deployment retention.ms = -1 raw_packets Raw frames · 300 GB retention alerts Alerts · 1 GB retention Kafka design points Decouples capture / parsing Parser restarts lose no data Producer retries = 5 Batch of 100 per send BufferError → flush and retry raw 300 GB / others 1 GB unified2 logs /var/log/suricata packet-inspector parse & detect engine · Consumer pulls raw_packets · pypkt (Cython) parses 70+ protocols · 5-tuple + MAC + protocol + payload → hash · Redis 10 s expiry window aggregates counts · Batch INSERT … RETURNING every 5 s · WhitelistInspector policy checks · Anomaly → Producer writes alerts · ThreadPool(16) + fail2restart self-healing Thread model in section 2 store.py · alert ingest Consume alerts → sec_alert table Syslog out · push to exercise range Supervisor daemons gunicorn · celery worker/beat store.py · unified2_recorder db_2_pilosa · packet_stat detect_device_offline cpulimit unified2_recorder idstools parse → deny-list alerts ingested Redis 6 DB0/1 Celery broker / results DB2 flow window TTL 10 s DB3 device activity TTL N min keyspace expiry events (Ex) PostgreSQL 10 packet_packet Flow records (with count / length) sec_alert_secalert Security alerts Assets · allow-list · deny-list · config wal_level = logical max_connections = 300 Kafka Connect · Debezium pgoutput logical replication → Avro auditor.public.* CDC stream · Schema Registry db_2_pilosa AvroConsumer writes bitmaps Pilosa bitmap index Traffic stats · allow-list self-learning Vue 2 frontend Vuex · Router · ECharts iView · axios · Electron Django 1.11 + DRF Gunicorn 8 workers :9003 Assets · allow-list · deny-list Security alerts · system alerts Packet audit · report export Policy backup / restore · settings Users · licence activation Allow-list self-learning Central-platform integration drf-yasg API docs About 12 Django apps Celery worker + beat Redis as broker Reports · policy backup · stats Expired-data cleanup · process checks bigdaddy device agent Change IP / time / reboot Configure capture NICs · push rules Firmware upgrade RSA + AES Certificates · licence activation External integration Central platform · SM4 (national crypto) Syslog · exercise-range push 1 2 3 4 5 6 7 8 9 10
pop map[key] Into the insert queue ① Kafka consumer threadpoll(0.01) raw_packets raw deque·1000 ② Parse + aggregate threadpypkt: protocol → hash key In-memory map + Redis DB2 New key: SETEX key 10 s Existing key: count++ / bytes+= Also SETEX src_ip → DB3 ③ Expiry listener thread__keyevent@2__:expired parsed deque·1000 ④ Insert + detect thread (every 5 s) Batch INSERT … RETURNING count InspectorManager → allow-list checks Anomaly → thread_pool.submit(send_alert) PostgreSQL packet_packet Kafka alerts (anomaly list) ⑤ Rule hot-reload thread (every 5 s) rules, assets, thresholds → share_data Django backend API Main thread watches with fail2restart Checks the 5 threads' futures every 5 s Thread error → log and re-submit PG disconnect → keep reconnecting Pool backlog > 500 → back off Bounded deques prevent memory growth Producer queue buffers 100k messages Failed alert sends logged via callback
Asset check · check_ip_and_mac Allow-list check · check_whitelists Independent detection paths No Yes Yes No No · bitmap 0 Yes Mismatch · bitmap 2 Match / no payload rule Parsed flow recordBoth checks run together IP in asset table?Network device (router) MAC skipped Illegal device access 2 IP-MAC binding onand actual MAC differs from asset? MAC address conflict 4 Asset check passedNo alert Matches any allow-list 5-tuple?IP / MAC / port range / L4 / protocol Abnormal communication behaviour 3 Industrial-protocol payload rulesfunction code / field value / value range Abnormal communication content 5 Normal passResult bitmap = 3 Suricata signature hitCODESYS, S7 and other ICS exploit signatures Deny-list alert 1 Redis DB3 device key expiredNo traffic from this IP for N minutes Device silent 30

System data flow

Abstract sketch · no internal details

  1. Switch mirror port

    Out-of-band, read-only

  2. Capture probe → Kafka

    Scapy · raw_packets kept 300 GB

  3. Protocol-parsing engine

    Cython · 27 industrial protocols

  4. Policy detection

    Assets · allow-list (bitmap) · Suricata deny-list

    ▲ My part

  5. PostgreSQL → Pilosa

    Flow detail · CDC bitmap index

  6. Django + Vue

    Alerts · policies · reports

    ▲ My part

Team & daily life

Photos are shared with colleagues' consent, or other people's faces are blurred

What I learned here

In industrial security I learned about common attack techniques and privilege escalation, met industrial protocols such as Modbus, S7COMM, IEC104 and OPC for the first time, and got more fluent with Django / DRF and Celery while writing business code.

Protocol parsing is CPU-bound, byte-level work: the parsing library compiles hot paths to C extensions with Cython and keeps a Python interface for upper layers. I also took part in a C++-to-Rust migration in this period. Redis expiry events as aggregation windows and bitmaps to tell anomaly types apart were simple primitives that absorbed the write and detection load of high-frequency control traffic.

TECH STACK

  • Python
  • Django / DRF
  • Celery
  • Kafka
  • Redis
  • PostgreSQL
  • Pilosa
  • Suricata
  • Cython
  • Vue 2
  • Docker