Community ID

#!/usr/bin/env python3
"""
Community ID implementation from scratch — to understand the spec.
For production use: pip install communityid
"""
import socket
import struct
import hmac
import hashlib
import base64

def community_id(src_ip: str, dst_ip: str,
                 src_port: int, dst_port: int,
                 proto: int, seed: int = 0) -> str:
    """
    Compute Community ID v1 for a flow.
    proto: 6=TCP, 17=UDP, 1=ICMP, 58=ICMPv6
    """
    src_addr = socket.inet_aton(src_ip)
    dst_addr = socket.inet_aton(dst_ip)

    # Normalize direction: lower IP is always "source"
    if src_addr > dst_addr:
        src_addr, dst_addr = dst_addr, src_addr
        src_port, dst_port = dst_port, src_port
    elif src_addr == dst_addr and src_port > dst_port:
        src_port, dst_port = dst_port, src_port

    # Build the hash input
    seed_bytes = struct.pack('>H', seed)      # 2 bytes, big-endian
    proto_byte = struct.pack('B', proto)       # 1 byte
    pad_byte = b'\x00'                         # 1 byte padding
    sp_bytes = struct.pack('>H', src_port)    # 2 bytes big-endian
    dp_bytes = struct.pack('>H', dst_port)    # 2 bytes big-endian

    data = seed_bytes + src_addr + dst_addr + proto_byte + pad_byte + sp_bytes + dp_bytes

    # HMAC-SHA1
    digest = hmac.new(seed_bytes, data, hashlib.sha1).digest()
    return '1:' + base64.b64encode(digest).decode()


# Examples
print(community_id('10.0.1.50', '185.220.101.47', 49812, 443, proto=6))
# → 1:hO+sN4H+MG5MY/8hIoss3QXS2+Iw=

# Verify that reversed direction gives the same hash
print(community_id('185.220.101.47', '10.0.1.50', 443, 49812, proto=6))
# → 1:hO+sN4H+MG5MY/8hIoss3QXS2+Iw=  (same — normalized)

# Using the communityid library (production use)
import communityid

cid = communityid.CommunityID()
print(cid.calc(communityid.FlowTuple.make_tcp('10.0.1.50', '185.220.101.47', 49812, 443)))

# Computing from a Zeek conn.log CSV row
import csv
with open('conn.log', 'r') as f:
    for row in csv.DictReader(f, delimiter='\t'):
        cid_hash = community_id(
            row['id.orig_h'], row['id.resp_h'],
            int(row['id.orig_p']), int(row['id.resp_p']),
            proto=6 if row['proto'] == 'tcp' else 17
        )
        print(f"{row['ts']}  {cid_hash}  {row['id.orig_h']}:{row['id.orig_p']}"
              f" → {row['id.resp_h']}:{row['id.resp_p']}")

# Scenario: Suricata fired an alert. Pivot to all related log sources.

# Step 1: Get the Community ID from the Suricata EVE alert
ALERT_CID=$(cat /var/log/suricata/eve.json | \
    jq -r 'select(.event_type=="alert" and
                  .alert.signature_id==2028465) | .community_id' | head -1)

echo "Community ID: $ALERT_CID"
# → 1:hO+sN4H+MG5MY/8hIoss3QXS2+Iw=

# Step 2: Find the matching Zeek conn.log entry
grep "$ALERT_CID" /var/log/zeek/current/conn.log | \
    awk '{print $1, $4, $5, $6, $7, $9, $10, $11}' # ts,src,dst,proto,service,duration,orig_bytes,resp_bytes

# Step 3: Find the TLS metadata (JA3, SNI, certificate)
grep "$ALERT_CID" /var/log/zeek/current/ssl.log | \
    awk '{print $8, $10, $11, $22, $23}' # ja3, server_name, subject, issuer

# Step 4: Find any file transfers in this session
grep "$ALERT_CID" /var/log/zeek/current/files.log | \
    awk '{print $4, $5, $8, $11}' # source, depth, mime_type, sha256

# Step 5: Search Arkime for the PCAP session
# Arkime REST API
curl -u analyst:changeme \
    "http://localhost:8005/api/sessions?query=communityId:$ALERT_CID" | jq '.data[0]'

# Step 6: In Elasticsearch — search all data sources at once
curl -XGET "http://elastic:9200/_search" -H 'Content-Type: application/json' -d "{
  \"query\": {
    \"term\": { \"network.community_id\": \"$ALERT_CID\" }
  }
}" | jq '.hits.hits[] | {_index, event_type: ._source.event_type, ts: ._source[\"@timestamp\"]}'