Documentation

XE009: Distributed Threat Intelligence Pipeline

Back to documentation index
---
{
  "@context": "https://nxdlang.org/schema",
  "doc_id": "XE009",
  "title": "",
  "description": "",
  "layer": "Examples",
  "category": "XL Examples",
  "keywords": [],
  "doc_version": "1.0",
  "status": "active"
}
---


# XE009: Distributed Threat Intelligence Pipeline
  (ingest, normalize, enrich, correlate, channels, multi‑module, supervision)

# Canonical NXD (XL‑Layer)

```nxd
MODULE ti.types

TYPE RESULT UNION { OK(any), ERR(string) }
TYPE OPTION UNION { SOME(any), NONE }

TYPE RAW_EVENT {
    SRC: string,
    PAYLOAD: string,
    TS: int
}

TYPE NORM_EVENT {
    SRC: string,
    TYPE: string,
    VALUE: string,
    TS: int
}

TYPE ENRICHED_EVENT {
    SRC: string,
    TYPE: string,
    VALUE: string,
    GEO: string,
    ASN: string,
    TS: int
}

TYPE ALERT {
    NAME: string,
    SEVERITY: string,
    MESSAGE: string,
    TS: int
}

TYPE CHANNEL<RAW_EVENT> { }
TYPE CHANNEL<NORM_EVENT> { }
TYPE CHANNEL<ENRICHED_EVENT> { }
TYPE CHANNEL<ALERT> { }
```



## MODULE: Ingest

```nxd
MODULE ti.ingest
IMPORT ti.types

FUNC EMIT(IN: CHANNEL<RAW_EVENT>, SRC: string, PAYLOAD: string):
    LET E SET RAW_EVENT {
        SRC: SRC,
        PAYLOAD: PAYLOAD,
        TS: NOW()
    }
    SEND E TO IN
```



## MODULE: Normalizer

```nxd
MODULE ti.normalize
IMPORT ti.types

FUNC PARSE(P: string): OPTION:
    # simple pattern: "type:value"
    IF NOT (P CONTAINS ":"):
        RETURN NONE

    LET PARTS SET SPLIT(P, ":")
    RETURN SOME({ TYPE: PARTS[0], VALUE: PARTS[1] })

FUNC NORMALIZE(IN: CHANNEL<RAW_EVENT>, OUT: CHANNEL<NORM_EVENT>):
    LOOP:
        LET R SET RECV IN
        LET P SET PARSE(R.PAYLOAD)

        MATCH P:
            CASE SOME(D):
                LET N SET NORM_EVENT {
                    SRC: R.SRC,
                    TYPE: D.TYPE,
                    VALUE: D.VALUE,
                    TS: R.TS
                }
                SEND N TO OUT
            CASE NONE:
                NONE
```



## MODULE: Enrichment

```nxd
MODULE ti.enrich
IMPORT ti.types

FUNC GEOLOOKUP(V: string): string:
    # fake geo lookup
    IF V CONTAINS "192.168":
        RETURN "internal"
    RETURN "unknown"

FUNC ASNLOOKUP(V: string): string:
    # fake ASN lookup
    IF V CONTAINS "192.168":
        RETURN "AS-LOCAL"
    RETURN "AS-EXT"

FUNC ENRICH(IN: CHANNEL<NORM_EVENT>, OUT: CHANNEL<ENRICHED_EVENT>):
    LOOP:
        LET N SET RECV IN
        LET G SET GEOLOOKUP(N.VALUE)
        LET A SET ASNLOOKUP(N.VALUE)

        LET E SET ENRICHED_EVENT {
            SRC: N.SRC,
            TYPE: N.TYPE,
            VALUE: N.VALUE,
            GEO: G,
            ASN: A,
            TS: N.TS
        }

        SEND E TO OUT
```



## MODULE: Correlation Window

```nxd
MODULE ti.window
IMPORT ti.types

TYPE WINDOW { ITEMS: LIST<ENRICHED_EVENT> }

FUNC NEW_WINDOW(): WINDOW:
    RETURN WINDOW { ITEMS: [] }

FUNC ADD(W: WINDOW, E: ENRICHED_EVENT):
    PUSH W.ITEMS, E

FUNC TRIM(W: WINDOW, AGE: int):
    LET NOW SET NOW()
    LET NEW SET []
    LOOP X IN W.ITEMS:
        IF NOW SUB X.TS LT AGE:
            PUSH NEW, X
    W.ITEMS SET NEW

FUNC FIND(W: WINDOW, TYPE: string): LIST<ENRICHED_EVENT>:
    LET OUT SET []
    LOOP X IN W.ITEMS:
        IF X.TYPE EQ TYPE:
            PUSH OUT, X
    RETURN OUT
```



## MODULE: Correlation Rules

```nxd
MODULE ti.rules
IMPORT ti.types
IMPORT ti.window

FUNC RULE_PORTSCAN(W: WINDOW): OPTION:
    LET HITS SET FIND(W, "port")
    IF LEN(HITS) GT 10:
        RETURN SOME(ALERT {
            NAME: "portscan",
            SEVERITY: "high",
            MESSAGE: "multiple port hits detected",
            TS: NOW()
        })
    RETURN NONE

FUNC RULE_INTERNAL_ANOMALY(W: WINDOW): OPTION:
    LET INT SET []
    LOOP X IN W.ITEMS:
        IF X.GEO EQ "internal":
            PUSH INT, X

    IF LEN(INT) GT 5:
        RETURN SOME(ALERT {
            NAME: "internal-anomaly",
            SEVERITY: "medium",
            MESSAGE: "excessive internal activity",
            TS: NOW()
        })

    RETURN NONE

FUNC RULE_ASN_MISMATCH(W: WINDOW): OPTION:
    LET EXT SET []
    LOOP X IN W.ITEMS:
        IF X.ASN EQ "AS-EXT":
            PUSH EXT, X

    IF LEN(EXT) GT 3:
        RETURN SOME(ALERT {
            NAME: "asn-mismatch",
            SEVERITY: "critical",
            MESSAGE: "external ASN activity detected",
            TS: NOW()
        })

    RETURN NONE
```



## MODULE: Correlator

```nxd
MODULE ti.correlate
IMPORT ti.types
IMPORT ti.window
IMPORT ti.rules

FUNC CORRELATE(IN: CHANNEL<ENRICHED_EVENT>, OUT: CHANNEL<ALERT>):
    LET W SET NEW_WINDOW()

    LOOP:
        LET E SET RECV IN
        ADD(W, E)
        TRIM(W, 5000)

        LET A1 SET RULE_PORTSCAN(W)
        LET A2 SET RULE_INTERNAL_ANOMALY(W)
        LET A3 SET RULE_ASN_MISMATCH(W)

        MATCH A1:
            CASE SOME(A): SEND A TO OUT
            CASE NONE: NONE

        MATCH A2:
            CASE SOME(A): SEND A TO OUT
            CASE NONE: NONE

        MATCH A3:
            CASE SOME(A): SEND A TO OUT
            CASE NONE: NONE
```



## MODULE: Sink

```nxd
MODULE ti.sink
IMPORT ti.types

FUNC PRINT_ALERTS(CH: CHANNEL<ALERT>):
    LOOP:
        LET A SET RECV CH
        PRINTLN("[ALERT] " ADD A.NAME ADD " (" ADD A.SEVERITY ADD ") " ADD A.MESSAGE)
```



## MODULE: App Main

```nxd
MODULE app.main
IMPORT ti.types
IMPORT ti.ingest
IMPORT ti.normalize
IMPORT ti.enrich
IMPORT ti.correlate
IMPORT ti.sink

FUNC MAIN():
    LET RAW SET CHANNEL<RAW_EVENT>()
    LET NORM SET CHANNEL<NORM_EVENT>()
    LET ENR SET CHANNEL<ENRICHED_EVENT>()
    LET OUT SET CHANNEL<ALERT>()

    SPAWN NORMALIZE(RAW, NORM)
    SPAWN ENRICH(NORM, ENR)
    SPAWN CORRELATE(ENR, OUT)
    SPAWN PRINT_ALERTS(OUT)

    # simulate threat intel events
    LOOP I FROM 1 TO 15:
        EMIT(RAW, "scanner", "port:192.168.1." ADD I)
        SLEEP(0.1)

    LOOP J FROM 1 TO 8:
        EMIT(RAW, "internal", "sys:192.168.1." ADD J)
        SLEEP(0.1)

    LOOP K FROM 1 TO 5:
        EMIT(RAW, "external", "port:8.8.8." ADD K)
        SLEEP(0.1)

    SLEEP(3)

    RETURN NONE
```



# What this XL example demonstrates

### Multi‑module threat intelligence architecture  
- Ingest  
- Normalize  
- Enrich  
- Window  
- Rules  
- Correlator  
- Sink  
- App main

### Full pipeline  
RAW → NORM → ENRICHED → CORRELATED → ALERTS

### Enrichment  
- Geo lookup  
- ASN lookup  
- Type/value extraction

### Correlation rules  
- Portscan detection  
- Internal anomaly detection  
- ASN mismatch detection  
- Multi‑rule evaluation

### Channels + processes  
- Fully asynchronous  
- Real distributed‑system semantics

### Result + Option  
- Optional rule matches  
- Safe evaluation  
- Pattern matching

This is a **full subsystem**, suitable for real NXD agent training and backend mapping.