Skip to content

README.md

Latest commit

 

History

History
211 lines (161 loc) · 7.48 KB

README.md

File metadata and controls

211 lines (161 loc) · 7.48 KB
Raw

jsoncef

jsoncef converts JSON logs — including Elastic Common Schema (ECS) events — into ArcSight Common Event Format (CEF). Point it at NDJSON from Elasticsearch/Logstash/Beats, a JSON-emitting device, or an application log, and it produces CEF lines ready for any SIEM that consumes CEF.

Looking for classic syslog conversion (RFC3164/RFC5424 dialects, journald, vendor syslog formats)? That lives in the sibling project syslogcef. jsoncef focuses on JSON sources, though it will still fall back to basic syslog parsing when a non-JSON line shows up in the stream.

Architecture

Features

  • ECS to CEF out of the box: nested ECS documents are flattened (source.ip, event.action, ...) and mapped to canonical CEF extension keys (src, spt, dst, act, suser, proto, ...). ECS events are auto-detected — no flags needed.
  • Severity derived from event.severity or log.level, falling back to syslog priority.
  • Additional vendor mappings (Cisco, Linux, F5, VMware) plus a generic default, selectable with --source.
  • Custom overrides via JSON/YAML mapping files (--mapping-file).
  • Streaming CLI: stdin/file input, --watch tail mode, worker threads, stats, and no runtime dependencies.
  • Malformed lines never abort a stream: they are emitted as CEF events tagged flexString1=parse_error (or fail fast with --strict).
  • CEF-spec-correct escaping of untrusted log content in header and extension fields, including newline neutralization so crafted messages cannot split or forge records.

Installation

pip install jsoncef

From source:

git clone https://github.com/allamiro/jsoncef.git
cd jsoncef
pip install .

Requires Python 3.10 or later. No runtime dependencies (PyYAML only if you use YAML mapping files).

Quickstart

Convert an ECS NDJSON export (auto-detected):

jsoncef --input samples/ecs.ndjson
CEF:0|JSONCEF|jsoncef|0.2.0|FW-1001|connection_denied|7|deviceVendor=JSONCEF deviceProduct=jsoncef deviceVersion=0.2.0 end=2026-08-11T09:15:00+00:00 msg=Connection denied src=10.1.1.1 spt=51515 suser=alice dst=203.0.113.9 dpt=443 dvchost=fw01.example.com proto=tcp act=connection_denied outcome=failure cat=network cs1Label=ruleName cs1=deny-outbound ...

Stream JSON events from stdin and print statistics:

kubectl logs my-app | jsoncef --format json --stats

Watch a file for new events and append CEF to an output file:

jsoncef --input /var/log/app/events.ndjson --output /var/log/cef/app.cef --watch

Force a specific mapping instead of auto-detection:

jsoncef --input firewall.ndjson --source cisco

Library usage

import json

from jsoncef import convert_line, from_json, select_mapping, to_cef
from jsoncef.mappings import get_mapping

ecs_event = {
    "@timestamp": "2026-08-11T09:15:00Z",
    "event": {"action": "user_login", "outcome": "success"},
    "source": {"ip": "192.0.2.44"},
    "user": {"name": "bob"},
    "host": {"name": "app01"},
    "message": "User logged in",
}

# One call per line (mapping auto-selected):
print(convert_line(json.dumps(ecs_event)))

# Or step by step with full control:
event = from_json(ecs_event)
mapping = get_mapping("ecs")            # or select_mapping(event) to auto-detect
print(to_cef(event, "Example", "Collector", "1.0", mapping))

CLI reference

Run jsoncef --help for the full option list. Key flags:

Option Description
--input, -i Input file, or - for stdin (default).
--output, -o Output file, or - for stdout (default).
--format {syslog,json} Force input format instead of auto detection.
--source Mapping: ecs, cisco, linux, f5, vmware, default. Omit to auto-detect ECS and use default otherwise.
--mapping-file JSON/YAML extension overrides merged into the mapping result.
--watch Tail the input file for new lines.
--workers N Convert lines in parallel worker threads.
--tz Europe/Berlin Default timezone for naive timestamps.
--strict Abort on parse errors instead of emitting tagged fallback events.
--stats Print processed/failed counters to stderr.
--vendor / --product / --version Override the CEF header device fields.

ECS field mapping

The ecs mapping flattens nested documents and translates common ECS fields to canonical CEF keys:

ECS CEF
source.ip / client.ip src
source.port / client.port spt
destination.ip / server.ip dst
destination.port / server.port dpt
source.user.name, user.name suser
destination.user.name duser
source.bytes / destination.bytes in / out
host.name, host.ip dvchost, dvc
event.action, event.outcome, event.category act, outcome, cat
event.code / event.id / event.dataset CEF signature id
network.transport proto
url.original, http.request.method, user_agent.original request, requestMethod, requestClientApplication
file.name, file.path, file.size, file.hash.md5 fname, filePath, fsize, fileHash
process.name, process.pid deviceProcessName, dvcpid
rule.name cs1 (labelled ruleName)
message msg

Severity comes from event.severity (clamped to 0-10), else log.level (info → 3, warning → 5, error → 7, critical → 9, ...), else the syslog priority when present.

Mapping architecture

Mappings translate parsed events into the CEF signature, name, severity and extension dictionary. Built-ins live under jsoncef.mappings:

  • ecs: Elastic Common Schema translation (see table above)
  • default: generic conversion preserving message, host and process info
  • cisco, linux, f5, vmware: vendor-tuned field extraction

Mappings conform to a small protocol (map(event) -> MappingResult), so custom mappings are plain Python objects. For lighter customization, --mapping-file merges JSON/YAML overrides into the mapping result; override values support Python format strings referencing event fields ({src}, {msg}, ...).

Performance

scripts/bench.py measures conversion throughput:

python scripts/bench.py samples/ecs.ndjson --lines 10000

Single-threaded on a commodity x86_64 Linux host (Python 3.12), the converter sustains roughly 27k syslog lines/s and 14k ECS JSON lines/s. Use --workers when CPU-bound mappings dominate the workload.

Sample data & rsyslog templates

Sample ECS events live in samples/ecs.ndjson and test fixtures under tests/data/. For rsyslog JSON output configuration examples, see RSYSLOG_TEMPLATES.md.

Security

Log content is treated as untrusted input. Header and extension values are escaped per the CEF specification and CR/LF are neutralized so crafted messages cannot forge header fields or split records. To report a vulnerability, see SECURITY.md — please do not open public issues for security reports.

Development

git clone https://github.com/allamiro/jsoncef.git
cd jsoncef
python -m venv .venv && source .venv/bin/activate
pip install -e .[dev]
pytest
ruff check src && black --check src tests && mypy src

Contributions are welcome — see CONTRIBUTING.md. Notable changes are tracked in CHANGELOG.md.

License

Apache License 2.0. See LICENSE. Copyright (c) Tamir Suliman.