Skip to content

Commit e047956

Browse files
hackgnarhackgnar
committed
added support for an elk stack SIEM
1 parent 204d930 commit e047956

4 files changed

Lines changed: 322 additions & 0 deletions

File tree

‎requirements.txt‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2,3 +2,4 @@ docker==7.1.0
22
psycopg2-binary
33
python-on-whales==0.74.0
44
rich>=13.0
5+
elasticsearch>=9.0

‎setc/modules/elasticsearch.py‎

Lines changed: 280 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,280 @@
1+
from __future__ import annotations
2+
3+
import json
4+
import logging
5+
import time
6+
import urllib.request
7+
import urllib.error
8+
9+
import docker
10+
import docker.models.containers
11+
from utils import prefixed_name
12+
13+
logger = logging.getLogger(__name__)
14+
15+
# Maps format directory names (as written under /data/<cve>/) to (index_name, is_json).
16+
INDEX_MAP = {
17+
"zeek": ("zeek", True),
18+
"cim": ("cim", True),
19+
"ecs": ("ecs", True),
20+
"ocsf": ("ocsf", True),
21+
"cef": ("cef", False),
22+
"udm": ("udm", True),
23+
}
24+
25+
26+
class ElasticsearchModule:
27+
"""Optional SIEM module that runs an Elasticsearch container and indexes SETC logs."""
28+
29+
def __init__(self, docker_client: docker.DockerClient, volume_name: str = "set_logs",
30+
network_name: str = "set_framework_net", elasticsearch_password: str = "password1234",
31+
prefix: str = "") -> None:
32+
"""Initialize Elasticsearch module with Docker client and connection settings."""
33+
self.client = docker_client
34+
self.volume = volume_name
35+
self.network = network_name
36+
self.prefix = prefix
37+
self.password = elasticsearch_password
38+
self.es_container = None
39+
self.kibana_container = None
40+
self._es_client = None
41+
self.setup_complete = False
42+
self._data_views_created = False
43+
44+
def _prefixed(self, name: str) -> str:
45+
"""Return the session-prefixed version of a container name."""
46+
return prefixed_name(self.prefix, name)
47+
48+
def _find_existing(self, image: str) -> docker.models.containers.Container | None:
49+
"""Find a running container for *image* already attached to our network."""
50+
for container in self.client.containers.list(all=True,
51+
filters={"ancestor": image}):
52+
mounts = container.attrs.get("Mounts", [])
53+
for m in mounts:
54+
if m.get("Name") == self.volume:
55+
if container.status != "running":
56+
logger.info("Starting stopped container: %s", container.name)
57+
container.start()
58+
return container
59+
# Kibana has no volume mount — match by network instead
60+
nets = container.attrs.get("NetworkSettings", {}).get("Networks", {})
61+
if self.network in nets:
62+
if container.status != "running":
63+
logger.info("Starting stopped container: %s", container.name)
64+
container.start()
65+
return container
66+
return None
67+
68+
def setup(self) -> None:
69+
"""Start Elasticsearch and Kibana containers, or reuse existing ones."""
70+
existing_es = self._find_existing("docker.elastic.co/elasticsearch/elasticsearch:9.0.0")
71+
if existing_es:
72+
logger.info("Reusing existing Elasticsearch container: %s", existing_es.name)
73+
self.es_container = existing_es
74+
self.setup_complete = True
75+
else:
76+
self.es_container = self.client.containers.run(
77+
"docker.elastic.co/elasticsearch/elasticsearch:9.0.0", detach=True,
78+
name=self._prefixed("elasticsearch"),
79+
volumes={self.volume: {"bind": "/data", "mode": "ro"}},
80+
ports={"9200/tcp": 9200},
81+
environment=[
82+
"discovery.type=single-node",
83+
"xpack.security.enabled=true",
84+
f"ELASTIC_PASSWORD={self.password}",
85+
"xpack.security.http.ssl.enabled=false",
86+
"ES_JAVA_OPTS=-Xms512m -Xmx512m",
87+
],
88+
network=self.network,
89+
)
90+
91+
existing_kb = self._find_existing("docker.elastic.co/kibana/kibana:9.0.0")
92+
if existing_kb:
93+
logger.info("Reusing existing Kibana container: %s", existing_kb.name)
94+
self.kibana_container = existing_kb
95+
else:
96+
es_name = self.es_container.name
97+
self.kibana_container = self.client.containers.run(
98+
"docker.elastic.co/kibana/kibana:9.0.0", detach=True,
99+
name=self._prefixed("kibana"),
100+
ports={"5601/tcp": 5601},
101+
environment=[
102+
f"ELASTICSEARCH_HOSTS=http://{es_name}:9200",
103+
"ELASTICSEARCH_USERNAME=kibana_system",
104+
f"ELASTICSEARCH_PASSWORD={self.password}",
105+
"NODE_OPTIONS=--max-old-space-size=512",
106+
],
107+
network=self.network,
108+
)
109+
110+
def is_ready(self) -> bool:
111+
"""Return True if the Elasticsearch container is accepting connections."""
112+
if not self.es_container:
113+
return False
114+
try:
115+
result = self.es_container.exec_run(
116+
["curl", "-s", "-u", f"elastic:{self.password}", "http://localhost:9200/_cluster/health"],
117+
demux=False,
118+
)
119+
return result.exit_code == 0
120+
except (docker.errors.NotFound, docker.errors.APIError):
121+
return False
122+
123+
def _get_client(self):
124+
"""Return an Elasticsearch client, creating one lazily."""
125+
if self._es_client is None:
126+
from elasticsearch import Elasticsearch
127+
self._es_client = Elasticsearch(
128+
"http://127.0.0.1:9200",
129+
basic_auth=("elastic", self.password),
130+
)
131+
return self._es_client
132+
133+
def post_setup(self) -> None:
134+
"""Set up the kibana_system user password and create indices for all log formats."""
135+
# Set the kibana_system password so Kibana can authenticate to Elasticsearch
136+
self.es_container.exec_run([
137+
"curl", "-s", "-X", "POST",
138+
"-u", f"elastic:{self.password}",
139+
"-H", "Content-Type: application/json",
140+
f"http://localhost:9200/_security/user/kibana_system/_password",
141+
"-d", json.dumps({"password": self.password}),
142+
])
143+
es = self._get_client()
144+
for index_name, _ in INDEX_MAP.values():
145+
es.indices.create(index=index_name, ignore=400)
146+
self.setup_complete = True
147+
logger.info("Elasticsearch indices created")
148+
149+
def _create_kibana_data_views(self) -> None:
150+
"""Create a Kibana data view for each index so logs are visible in Discover."""
151+
import base64
152+
auth = base64.b64encode(f"elastic:{self.password}".encode()).decode()
153+
headers = {
154+
"kbn-xsrf": "true",
155+
"Content-Type": "application/json",
156+
"Authorization": f"Basic {auth}",
157+
}
158+
for index_name, _ in INDEX_MAP.values():
159+
body = json.dumps({
160+
"data_view": {
161+
"title": index_name,
162+
"name": index_name,
163+
},
164+
}).encode()
165+
req = urllib.request.Request(
166+
"http://127.0.0.1:5601/api/data_views/data_view",
167+
data=body, headers=headers, method="POST",
168+
)
169+
try:
170+
urllib.request.urlopen(req)
171+
logger.debug("Created Kibana data view: %s", index_name)
172+
except urllib.error.HTTPError as e:
173+
if e.code == 409:
174+
logger.debug("Kibana data view already exists: %s", index_name)
175+
else:
176+
logger.warning("Failed to create Kibana data view %s: %s", index_name, e)
177+
178+
def ingest_logs(self, cve_name: str) -> None:
179+
"""Read log files from the volume and index them into Elasticsearch."""
180+
es = self._get_client()
181+
182+
for fmt_dir, (index_name, is_json) in INDEX_MAP.items():
183+
# List files in /data/<cve_name>/<format>/
184+
base_path = f"/data/{cve_name}/{fmt_dir}"
185+
try:
186+
ls_result = self.es_container.exec_run(["ls", base_path], demux=False)
187+
if ls_result.exit_code != 0:
188+
continue
189+
filenames = ls_result.output.decode().strip().split("\n")
190+
except (docker.errors.NotFound, docker.errors.APIError):
191+
continue
192+
193+
for filename in filenames:
194+
if not filename.strip():
195+
continue
196+
filepath = f"{base_path}/{filename.strip()}"
197+
try:
198+
cat_result = self.es_container.exec_run(["cat", filepath], demux=False)
199+
if cat_result.exit_code != 0:
200+
continue
201+
content = cat_result.output.decode()
202+
except (docker.errors.NotFound, docker.errors.APIError):
203+
continue
204+
205+
# Derive log_type from filename (e.g. "cim_http.log" -> "http")
206+
log_type = filename.strip().rsplit(".", 1)[0] # remove .log extension
207+
# Strip format prefix if present (e.g. "cim_http" -> "http")
208+
for pfx in ("cim_", "ecs_", "ocsf_", "cef_", "udm_", "zeek_"):
209+
if log_type.startswith(pfx):
210+
log_type = log_type[len(pfx):]
211+
break
212+
213+
if is_json:
214+
self._ingest_json(es, index_name, cve_name, log_type, content)
215+
else:
216+
self._ingest_text(es, index_name, cve_name, log_type, content)
217+
218+
logger.info("Elasticsearch ingest complete for %s", cve_name)
219+
220+
if not self._data_views_created and self.kibana_container:
221+
self._create_kibana_data_views()
222+
self._data_views_created = True
223+
224+
def _ingest_json(self, es, index_name: str, cve_name: str, log_type: str,
225+
content: str) -> None:
226+
"""Parse JSON content and bulk-index each event."""
227+
from elasticsearch import helpers
228+
try:
229+
data = json.loads(content)
230+
except json.JSONDecodeError:
231+
logger.debug("Skipping non-JSON content in %s", index_name)
232+
return
233+
events = data if isinstance(data, list) else [data]
234+
actions = []
235+
for event in events:
236+
doc = {"cve_name": cve_name, "log_type": log_type}
237+
doc.update(event)
238+
actions.append({"_index": index_name, "_source": doc})
239+
if actions:
240+
helpers.bulk(es, actions)
241+
242+
def _ingest_text(self, es, index_name: str, cve_name: str, log_type: str,
243+
content: str) -> None:
244+
"""Index each non-empty line as a text event (used for CEF)."""
245+
from elasticsearch import helpers
246+
actions = []
247+
for line in content.strip().split("\n"):
248+
line = line.strip()
249+
if not line:
250+
continue
251+
actions.append({
252+
"_index": index_name,
253+
"_source": {"cve_name": cve_name, "log_type": log_type, "event": line},
254+
})
255+
if actions:
256+
helpers.bulk(es, actions)
257+
258+
def cleanup(self, remove: bool = False) -> None:
259+
"""Optionally stop and remove the Elasticsearch and Kibana containers, or leave them running."""
260+
if self._es_client:
261+
try:
262+
self._es_client.close()
263+
except Exception:
264+
pass
265+
for container, label in [(self.kibana_container, "Kibana"),
266+
(self.es_container, "Elasticsearch")]:
267+
if not container:
268+
continue
269+
if remove:
270+
try:
271+
container.stop()
272+
container.remove()
273+
logger.info("%s container removed", label)
274+
except (docker.errors.NotFound, docker.errors.APIError) as e:
275+
logger.warning("Could not remove %s container: %s", label, e)
276+
if not remove:
277+
if self.es_container:
278+
logger.info("Elasticsearch available at http://localhost:9200 (user: elastic)")
279+
if self.kibana_container:
280+
logger.info("Kibana available at http://localhost:5601 (user: elastic)")

‎setc/setc.py‎

Lines changed: 22 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -13,6 +13,7 @@
1313
from modules.zeek import ZeekModule
1414
from modules.splunk import SplunkModule
1515
from modules.postgres import PostgresModule
16+
from modules.elasticsearch import ElasticsearchModule
1617
from modules.docker_process_logger import DockerProcessLogs
1718

1819

@@ -136,6 +137,9 @@ def parse_args() -> tuple[argparse.Namespace, list[dict[str, Any]]]:
136137
module_group.add_argument("--postgres",
137138
help="Create a PostgreSQL instance and populate it with SETC logs.",
138139
action='store_true')
140+
module_group.add_argument("--elk",
141+
help="Create an Elasticsearch instance with Kibana and populate it with SETC logs. Kibana UI available at http://localhost:5601.",
142+
action='store_true')
139143
module_group.add_argument("--no-zeek",
140144
help="Disable zeek pcap log parsing (enabled by default).",
141145
action='store_true')
@@ -153,6 +157,9 @@ def parse_args() -> tuple[argparse.Namespace, list[dict[str, Any]]]:
153157
cleanup_group.add_argument("--cleanup_postgres",
154158
help="Remove the PostgreSQL container after SETC completes.",
155159
action='store_true')
160+
cleanup_group.add_argument("--cleanup_elk",
161+
help="Remove the Elasticsearch and Kibana containers after SETC completes.",
162+
action='store_true')
156163

157164
args = parser.parse_args()
158165

@@ -234,6 +241,7 @@ def main() -> None:
234241
zeek = None
235242
splunk = None
236243
postgres = None
244+
elasticsearch = None
237245

238246
try:
239247
########################################
@@ -292,6 +300,12 @@ def main() -> None:
292300
postgres_password=args.password, prefix=prefix)
293301
postgres.setup()
294302

303+
if args.elk:
304+
elasticsearch = ElasticsearchModule(client, volume_name=args.volume,
305+
network_name=args.network,
306+
elasticsearch_password=args.password, prefix=prefix)
307+
elasticsearch.setup()
308+
295309

296310
########################################
297311
### Post UP Core Modules ###
@@ -423,6 +437,12 @@ def main() -> None:
423437
if postgres and postgres.setup_complete:
424438
postgres.ingest_logs(system_config["name"])
425439

440+
if elasticsearch and elasticsearch.is_ready() and not elasticsearch.setup_complete:
441+
logger.info("Creating Elasticsearch indices")
442+
elasticsearch.post_setup()
443+
if elasticsearch and elasticsearch.setup_complete:
444+
elasticsearch.ingest_logs(system_config["name"])
445+
426446
except Exception as e:
427447
logger.error("Error processing system %s: %s", system_config['name'], e)
428448
logger.warning("Continuing to next system...")
@@ -455,6 +475,8 @@ def main() -> None:
455475
splunk.cleanup(remove=args.cleanup_splunk)
456476
if postgres:
457477
postgres.cleanup(remove=args.cleanup_postgres)
478+
if elasticsearch:
479+
elasticsearch.cleanup(remove=args.cleanup_elk)
458480

459481

460482
if __name__ == "__main__":

‎tests/test_unit.py‎

Lines changed: 19 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -29,6 +29,7 @@
2929
)
3030
from utils import prefixed_name # noqa: E402
3131
from modules.postgres import TABLES, FORMAT_TABLE # noqa: E402
32+
from modules.elasticsearch import INDEX_MAP # noqa: E402
3233

3334
# ---------------------------------------------------------------------------
3435
# Sample data
@@ -594,3 +595,21 @@ def test_format_table_mapping(self):
594595
else:
595596
assert is_json is True
596597
assert table_name == f"{fmt_dir}_logs"
598+
599+
600+
# ===================================================================
601+
# 10. ElasticsearchModule constants
602+
# ===================================================================
603+
class TestElasticsearchModule:
604+
"""Tests for ElasticsearchModule index/format constants."""
605+
606+
def test_index_map_has_all_formats(self):
607+
expected = {"zeek", "cim", "ecs", "ocsf", "cef", "udm"}
608+
assert set(INDEX_MAP.keys()) == expected
609+
610+
def test_cef_is_non_json(self):
611+
_, is_json = INDEX_MAP["cef"]
612+
assert is_json is False
613+
for fmt_dir, (index_name, is_json) in INDEX_MAP.items():
614+
if fmt_dir != "cef":
615+
assert is_json is True, f"{fmt_dir} should be JSON"

0 commit comments

Comments
 (0)