! pmacct configuration for IPFIX collection
daemonize: false
pcap_interface: eth0
 
! IPFIX/NetFlow listener
nfacctd_port: 9995
nfacctd_ip: 0.0.0.0
 
! Output to Kafka
plugins: kafka
 
! Kafka configuration
kafka_output: json
kafka_topic: network-flows
kafka_broker_host: kafka-1:9092,kafka-2:9092,kafka-3:9092
kafka_partition_by: src_ip
 
! Fields to capture
aggregate: src_host, dst_host, src_port, dst_port, proto, \
           bytes, packets, tos, timestamp_start, timestamp_end

{
  "event_type": "flow",
  "src_ip": "10.1.50.23",
  "dst_ip": "192.168.1.100",
  "src_port": 52341,
  "dst_port": 22,
  "proto": "tcp",
  "bytes": 4523,
  "packets": 42,
  "timestamp_start": "2024-12-05T14:23:01Z",
  "timestamp_end": "2024-12-05T14:23:58Z"
}

kafka-topics.sh --create \
  --topic network-flows \
  --partitions 24 \
  --replication-factor 3 \
  --config retention.ms=21600000 \
  --config retention.bytes=107374182400

CREATE TABLE network_flows
(
    timestamp DateTime64(3),
    src_ip IPv4,
    dst_ip IPv4,
    src_port UInt16,
    dst_port UInt16,
    proto Enum8('tcp' = 6, 'udp' = 17, 'icmp' = 1),
    bytes UInt64,
    packets UInt32,
 
    -- Materialized columns for faster queries
    date Date MATERIALIZED toDate(timestamp),
    hour UInt8 MATERIALIZED toHour(timestamp)
)
ENGINE = MergeTree()
PARTITION BY toYYYYMM(timestamp)
ORDER BY (dst_port, src_ip, dst_ip, timestamp)
TTL timestamp + INTERVAL 6 MONTH
SETTINGS index_granularity = 8192;

CREATE TABLE network_flows_queue
(
    timestamp DateTime64(3),
    src_ip IPv4,
    dst_ip IPv4,
    src_port UInt16,
    dst_port UInt16,
    proto String,
    bytes UInt64,
    packets UInt32
)
ENGINE = Kafka()
SETTINGS
    kafka_broker_list = 'kafka-1:9092,kafka-2:9092,kafka-3:9092',
    kafka_topic_list = 'network-flows',
    kafka_group_name = 'clickhouse-flows',
    kafka_format = 'JSONEachRow',
    kafka_num_consumers = 4;
 
CREATE MATERIALIZED VIEW network_flows_mv TO network_flows AS
SELECT
    timestamp,
    src_ip,
    dst_ip,
    src_port,
    dst_port,
    CAST(proto AS Enum8('tcp' = 6, 'udp' = 17, 'icmp' = 1)) as proto,
    bytes,
    packets
FROM network_flows_queue;

-- All unique src_ip/dst_ip pairs with SSH (port 22) in last 24 hours
SELECT DISTINCT
    src_ip,
    dst_ip
FROM network_flows
WHERE dst_port = 22
  AND timestamp >= now() - INTERVAL 24 HOUR

WITH recent_ssh AS (
    -- SSH pairs from last 24 hours
    SELECT DISTINCT
        src_ip,
        dst_ip
    FROM network_flows
    WHERE dst_port = 22
      AND timestamp >= now() - INTERVAL 24 HOUR
),
historical_ssh AS (
    -- SSH pairs from 3 months ago to 24 hours ago
    SELECT DISTINCT
        src_ip,
        dst_ip
    FROM network_flows
    WHERE dst_port = 22
      AND timestamp >= now() - INTERVAL 3 MONTH
      AND timestamp < now() - INTERVAL 24 HOUR
)
SELECT
    r.src_ip,
    r.dst_ip,
    'NEW' as status
FROM recent_ssh r
LEFT JOIN historical_ssh h
    ON r.src_ip = h.src_ip AND r.dst_ip = h.dst_ip
WHERE h.src_ip IS NULL
ORDER BY r.src_ip, r.dst_ip;

WITH recent_ssh AS (
    SELECT DISTINCT
        src_ip,
        dst_ip
    FROM network_flows
    WHERE dst_port = 22
      AND timestamp >= now() - INTERVAL 24 HOUR
),
historical_ssh AS (
    SELECT DISTINCT
        src_ip,
        dst_ip
    FROM network_flows
    WHERE dst_port = 22
      AND timestamp >= now() - INTERVAL 3 MONTH
      AND timestamp < now() - INTERVAL 24 HOUR
),
new_pairs AS (
    SELECT r.src_ip, r.dst_ip
    FROM recent_ssh r
    LEFT JOIN historical_ssh h
        ON r.src_ip = h.src_ip AND r.dst_ip = h.dst_ip
    WHERE h.src_ip IS NULL
)
SELECT
    np.src_ip,
    np.dst_ip,
    min(nf.timestamp) as first_seen,
    max(nf.timestamp) as last_seen,
    count(*) as connection_count,
    sum(nf.bytes) as total_bytes,
    sum(nf.packets) as total_packets
FROM new_pairs np
JOIN network_flows nf
    ON np.src_ip = nf.src_ip
   AND np.dst_ip = nf.dst_ip
   AND nf.dst_port = 22
   AND nf.timestamp >= now() - INTERVAL 24 HOUR
GROUP BY np.src_ip, np.dst_ip
ORDER BY first_seen DESC;

import clickhouse_connect
from datetime import datetime
 
client = clickhouse_connect.get_client(host='clickhouse')
 
new_ssh_pairs = client.query("""
    WITH recent_ssh AS (...),  -- query from above
    ...
""").result_rows
 
if new_ssh_pairs:
    send_alert(
        channel="#security-alerts",
        message=f"Found {len(new_ssh_pairs)} new SSH relationships",
        details=new_ssh_pairs
    )

-- New database connections (port 3306/5432)
WHERE dst_port IN (3306, 5432)
 
-- New SMB connections (port 445)
WHERE dst_port = 445
 
-- New connections to external IPs (cast to string for isIPAddressInRange)
WHERE NOT isIPAddressInRange(toString(dst_ip), '10.0.0.0/8')
  AND NOT isIPAddressInRange(toString(dst_ip), '172.16.0.0/12')
  AND NOT isIPAddressInRange(toString(dst_ip), '192.168.0.0/16')
 
-- New high-volume relationships (data exfiltration?)
HAVING total_bytes > 100000000  -- 100MB+

-- Hosts allowed to transfer large amounts of data externally
WITH allowed_large_transfers AS (
    SELECT arrayJoin([
        toIPv4('10.1.1.50'),   -- Backup server
        toIPv4('10.1.1.51'),   -- SFTP gateway
        toIPv4('10.1.2.10'),   -- Data sync server
        toIPv4('10.1.5.100')   -- Cloud sync agent
    ]) as ip
)
SELECT
    src_ip,
    dst_ip,
    sum(bytes) as total_bytes,
    count(*) as connection_count,
    min(timestamp) as first_seen,
    max(timestamp) as last_seen
FROM network_flows
WHERE timestamp >= now() - INTERVAL 24 HOUR
  AND src_ip BETWEEN toIPv4('10.0.0.0') AND toIPv4('10.255.255.255')
  AND NOT (dst_ip BETWEEN toIPv4('10.0.0.0') AND toIPv4('10.255.255.255'))
  AND src_ip NOT IN (SELECT ip FROM allowed_large_transfers)
GROUP BY src_ip, dst_ip
HAVING total_bytes > 100000000  -- 100MB+
ORDER BY total_bytes DESC
LIMIT 50

-- Security scanners and tools allowed to perform port scans
WITH allowed_scanners AS (
    SELECT arrayJoin([
        toIPv4('10.100.1.10'),  -- Nessus scanner
        toIPv4('10.100.1.11'),  -- Qualys appliance
        toIPv4('10.100.1.20'),  -- IT admin workstation
        toIPv4('10.100.2.5')    -- Ansible/config mgmt
    ]) as ip
)
SELECT
    src_ip,
    dst_ip,
    count(DISTINCT dst_port) as unique_ports,
    min(timestamp) as first_seen,
    max(timestamp) as last_seen
FROM network_flows
WHERE timestamp >= now() - INTERVAL 1 HOUR
  AND src_ip NOT IN (SELECT ip FROM allowed_scanners)
GROUP BY src_ip, dst_ip
HAVING unique_ports >= 20
ORDER BY unique_ports DESC

-- Monitoring systems that beacon legitimately
WITH allowed_beacons AS (
    SELECT ip, port FROM (
        SELECT arrayJoin([
            (toIPv4('10.50.1.10'), toUInt16(443)),   -- Nagios/Zabbix
            (toIPv4('10.50.1.11'), toUInt16(443)),   -- Prometheus
            (toIPv4('10.50.1.20'), toUInt16(80)),    -- Uptime monitor
            (toIPv4('10.50.2.5'),  toUInt16(8443))   -- Pingdom agent
        ]) as t
    ) ARRAY JOIN t.1 as ip, t.2 as port
),
-- Known external services with regular check-ins
allowed_destinations AS (
    SELECT arrayJoin([
        toIPv4('13.107.4.50'),    -- Microsoft 365 health
        toIPv4('172.217.14.78'),  -- Google APIs
        toIPv4('52.94.236.0')     -- AWS health checks
    ]) as ip
),
connections AS (
    SELECT
        src_ip,
        dst_ip,
        dst_port,
        timestamp,
        dateDiff('second',
            lagInFrame(timestamp) OVER (
                PARTITION BY src_ip, dst_ip, dst_port
                ORDER BY timestamp
            ),
            timestamp
        ) as interval_seconds
    FROM network_flows
    WHERE timestamp >= now() - INTERVAL 24 HOUR
      AND src_ip NOT IN (SELECT ip FROM allowed_beacons)
      AND dst_ip NOT IN (SELECT ip FROM allowed_destinations)
)
SELECT
    src_ip,
    dst_ip,
    dst_port,
    count(*) as beacon_count,
    round(avg(interval_seconds)) as avg_interval_sec,
    round(stddevPop(interval_seconds), 2) as interval_stddev
FROM connections
WHERE interval_seconds IS NOT NULL AND interval_seconds > 0
GROUP BY src_ip, dst_ip, dst_port
HAVING
    beacon_count >= 10                        -- At least 10 connections
    AND avg_interval_sec BETWEEN 60 AND 3600  -- 1min to 1hr intervals
    AND interval_stddev < avg_interval_sec * 0.1  -- Very consistent timing
ORDER BY beacon_count DESC
LIMIT 50

CREATE TABLE flow_allowlists (
    list_type Enum8('scanner' = 1, 'backup' = 2, 'monitor' = 3),
    ip IPv4,
    port UInt16 DEFAULT 0,
    description String,
    added_by String,
    added_at DateTime DEFAULT now()
) ENGINE = MergeTree()
ORDER BY (list_type, ip);
 
-- Add entries
INSERT INTO flow_allowlists VALUES
    ('scanner', '10.100.1.10', 0, 'Nessus vulnerability scanner', 'security-team', now()),
    ('backup', '10.1.1.50', 0, 'Primary backup server', 'infra-team', now()),
    ('monitor', '10.50.1.10', 443, 'Nagios server', 'ops-team', now());
 
-- Use in queries
WITH allowed AS (
    SELECT ip FROM flow_allowlists WHERE list_type = 'scanner'
)
SELECT ... WHERE src_ip NOT IN (SELECT ip FROM allowed)