Event Pipeline Design

This document describes how Cowrie events travel from the code that observes attacker activity to the output plugins: a session-scoped event log feeding a single central dispatcher. The event schema (the event ids and their attributes) is documented in Output Event Code Reference.

Terminology

event

A dictionary describing one observed action, carrying an eventid (e.g. cowrie.session.file_download), a human-readable message or format, and event-specific attributes.

session

The unique identifier of one attacker connection (transportId, e.g. ce8abc71b984). This is the key output consumers correlate on.

channel

One terminal, exec, subsystem, or forwarding stream within a connection. SSH multiplexes: a single transport can carry several channels, concurrently or in sequence. Telnet has exactly one channel per connection.

output plugin

A subclass of cowrie.core.output.Output (jsonlog, mysql, s3, …) that receives finished events through its write() method.

Two streams

Cowrie produces two fundamentally different kinds of output:

diagnostic logging

Everything Cowrie has to say about itself: debug statements, development traces, protocol negotiation detail, factories starting and stopping, tracebacks, plugin trouble. Its audience is the developer and the operator; its semantics are log semantics (levels, verbosity, best-effort, human-formatted). This stream is by nature the larger one: most of what it carries never becomes an event. It stays on Twisted’s logging API and reaches cowrie.log.

events

The curated subset: structured records of attacker behavior (logins, commands, downloads) whose audience is downstream consumers – databases, SIEMs, jsonlog. Their semantics are data semantics: guaranteed attribution, stable schema, reliable delivery to every configured sink. These travel the event pipeline described here.

The two streams are separate all the way down. An event is dispatched exactly once, to the output plugins and to a built-in console sink that renders its human-readable line into cowrie.log; the event path has no dependency on the logging system. cowrie.log remains the superset it always was: all diagnostics, plus one rendered line per event, merged chronologically.

Architecture

Two pieces in cowrie.core.events carry every event from emission to the sinks:

command / protocol / transport                 diagnostics
  |                                              |
  |  self.protocol.events.dispatch(...)          |  log.msg(...)
  v                                              v
EventLog (identity bound at connectionMade)    Twisted log --> cowrie.log
  |                                              ^
  v                                              |
EventDispatcher -- enrich once                   |
  |                                              |
  +--> plugin.write()  x N, error-isolated       |
  +--> console renderer -- one line per event ---+

EventLog – session-scoped emitter

A small object created by the transport in connectionMade(), carrying the session’s identity once:

class EventLog:
    """Emits events for one attacker connection, identity bound at
    creation."""

    def __init__(self, dispatcher, **identity):
        ...

    def dispatch(self, eventid: str, fmt: str, **fields) -> None:
        """Stamp the bound identity, hand the event to the dispatcher."""

The transport owns it (self.events = EventLog(...)); the protocol and commands reach it through the objects they already hold (self.protocol.events), and connectionLost leaves the reference in place. Because identity is bound at creation, an event is attributable from any execution context. A download command’s deferred callbacks close over the command instance, so a transfer that completes after connectionLost still holds the EventLog and its event arrives fully attributed – there is no downstream table that could have forgotten the session.

The bound identity is authoritative: session, src_ip, and protocol from the identity always win over same-named fields passed to dispatch(). This matters because some events carry attacker-influenced values under those names – a direct-tcpip request’s claimed originator, for instance – which must not masquerade as the connection’s real source.

Transports construct the EventLog and emit the connection event through one helper:

self.events = transport_events(
    self.factory, self.transport,
    session=self.transportId, protocol="ssh", src_ip=src_ip,
)

transport_events() returns the emitter (or None when the running application provides no dispatcher, e.g. in some tests) and dispatches cowrie.session.connect from the bound identity. Symmetrically, EventLog.session_closed(duration_ms) dispatches cowrie.session.closed and marks the emitter closed. Keeping both in the EventLog means the four transports (ssh, telnet, and the two proxy frontends) do not each carry a copy of the connection event’s id and format string.

Late events

Events dispatched after the connection has closed carry an additional late: true attribute so consumers can distinguish them. late refers to the connection having closed, not a channel: a channel ending mid-session is not late.

Late events exist only while some live object (a pending deferred’s command instance) still references the session’s EventLog; when the last reference is collected, no further events can be emitted for that session. The bound is therefore the lifetime of in-flight work, not wall time – an attacker cannot keep a closed session emitting indefinitely without also keeping a transfer open, which existing timeouts already bound (treq timeout=10, TFTP retry caps).

Channels

EventLog.child(**extra) returns a lightweight view sharing the dispatcher and the bound connection identity, overlaying extra fields – the same idea as a structlog bind() – and following the parent connection’s closed state. It is the mechanism by which a channel could add a channel attribute to every event it dispatches, distinguishing concurrent SSH channels in the event stream the way the per-channel ttylog and stdin artifacts already do. The channels do not currently derive a child emitter; they share the transport’s root EventLog. The capability is in place for when per-channel attribution is wired.

EventDispatcher – single fan-out

One instance, owned by the application container (the tac), holding the list of sinks – the configured output plugins plus the console renderer:

class EventDispatcher:
    def dispatch(self, event: dict) -> None:
        """Enrich once (sensor, uuid, timestamp, message interpolation,
        bytes conversion), then deliver to each sink, isolating failures
        per sink."""

Enrichment runs once per event. Each sink receives its own copy so a plugin that mutates the event it is handed cannot corrupt what the sinks after it see. Sink exceptions are caught and logged (rate-limited per sink), and do not affect other sinks or the emitting session. The dispatcher keeps counters – events dispatched, events dropped after stop, per-sink failures – as the natural hook for pipeline observability.

Session-less events

Most events carry a session; the output plugins’ write() assumes it. A few events are operational rather than session-scoped (abuseipdb rate-limit notices, the plugin-started banner). The dispatcher routes an event with no session only to sinks that opt in by setting accepts_sessionless = True – the console renderer does. A session-scoped sink never sees a session-less event and so never has to guard for the missing key.

Output plugins as event sources

Output plugins are not only sinks. virustotal, reversedns, and greynoise emit session-attributed enrichment events (cowrie.virustotal.scanfile, cowrie.reversedns.connect, …) and abuseipdb emits session-less operational events. These are events, not diagnostics, and must reach every configured sink. Plugins therefore hold a reference to the EventDispatcher (set on the Output base class before any plugin loads, so an event dispatched from a plugin’s start() is delivered) and emit through Output.dispatch(), carrying the attribution of the event that triggered them, into the same fan-out as everything else.

The console renderer – events back into cowrie.log

Developers and operators read one merged, chronological cowrie.log in which event lines sit among the (far more numerous) diagnostic lines. A small built-in sink renders each event’s message into that log, stamping the session’s protocol,session,src_ip prefix explicitly. Rendered from the attributed event, a line carries its session no matter where the emitting code ran – a download callback’s lines no longer appear under [HTTP11ClientProtocol,client], making concurrent sessions distinguishable in the log. Rendering is thereby just another consumer – jsonlog renders to JSON, the console renderer renders to the diagnostic log – rather than something the emitter does as a side effect.

Calling patterns

A shell emitter such as command input, attributed by the bound identity and delivered from any execution context:

self.protocol.events.dispatch("cowrie.command.input", "CMD: %(input)s",
                              input=line)

A download command’s completion callback. If this fires after the attacker disconnected, the event arrives with late: true and full attribution, and the console renderer prints it under the session’s log prefix:

def collectioncomplete(self, data: None) -> None:
    ...
    self.protocol.events.dispatch(
        "cowrie.session.file_download",
        "Downloaded URL (%(url)s) with SHA-256 %(shasum)s to %(outfile)s",
        url=self.url.decode(),
        outfile=self.artifact.shasumFilename,
        shasum=self.artifact.shasum,
        duplicate=self.artifact.duplicate,
    )

A transport-level event, e.g. the SSH client version string:

self.events.dispatch(
    "cowrie.client.version",
    "Remote SSH version: %(version)s",
    version=self.otherVersionString,
)

An output plugin emitting an enrichment event for the event it is processing:

self.dispatch(
    eventid="cowrie.virustotal.scanfile",
    format="VT: New file %(sha256)s",
    session=event["session"],
    src_ip=event["src_ip"],
    protocol=event["protocol"],
    sha256=event["shasum"],
)

Login events attach the session’s emitter to the credential object rather than reaching a transport directly, since the credential is what the auth checker receives:

creds = UsernamePasswordIP(username, password, src_ip, events=self.events)

For public-key authentication Twisted constructs the credential internally, so the SSH userauth service wraps the portal for that one call (EventsAttachingPortal) and attaches the transport’s emitter to whatever credential passes through, letting the public-key checker dispatch attributed events.

Operational considerations

No configuration changes

Plugin selection, loading, and configuration are untouched; cowrie_plugin.py builds the same plugin list and hands it to the dispatcher. An upgrade is a restart.

Plugin failure handling

write() exceptions are caught per plugin and rate-limited (first failure, then periodic summaries), so an unreachable database does not turn every attacker keystroke into an error line and flood cowrie.log. One plugin raising does not stop delivery to the others.

Shutdown

The dispatcher and each plugin’s stop() run at after-shutdown, so the final events of connections torn down during shutdown (cowrie.session.closed, the ttylog-closed record) deliver before a sink’s backend (a database pool, an hpfeeds client) is closed. The dispatcher tolerates dispatch-after-stop (drop and count) so a late deferred firing during teardown cannot raise into the reactor.

Blocking plugins

write() is called synchronously in the reactor thread – a plugin doing blocking I/O stalls the honeypot. The dispatcher is the single place a delivery queue or thread pool could be added later.

Pipeline observability

The dispatcher’s counters (events dispatched, dropped, per-sink failures) are the natural point to expose delivery health – turning “is my ELK feed complete?” from guesswork into a metric. They are maintained but not yet surfaced to a plugin or command.

Log volume

cowrie.log content is unchanged: one console line per event, plus diagnostics as before. Plugins no longer scan every Twisted log line, so per-event plugin work is (events x N plugins), not (all log lines x N plugins).

Security considerations

Audit-trail integrity

Per-plugin error isolation means a plugin crash – including one triggered by crafted attacker input – cannot suppress delivery of that event to the remaining plugins. There is no per-plugin session table left to desynchronize.

Attacker-controlled data

Event fields (commands, URLs, filenames, version strings) are attacker input, and the bound identity overrides same-named fields so an attacker cannot spoof session/src_ip/protocol in the structured output. The dispatcher’s enrichment step is the single choke point where console rendering could additionally strip control characters and escape sequences (log-injection via \n or ANSI codes) while the structured event keeps the raw bytes for forensics; that stripping is a candidate for future work.

Event floods

An attacker spamming commands generates events at line rate, unchanged from before. The dispatcher’s counters provide a detection hook; rate-limiting event generation stays with the emitting subsystems (e.g. the download rate limiter).

Testing

EventLog and EventDispatcher are plain objects with no dependency on the global Twisted log, so unit tests inject a capturing fake sink and assert on delivered dictionaries – no log-observer fixtures, no regex setup. The helpers in cowrie.test.eventcapture (capture_eventlog, capture_events, events_of, make_exec_transport) wire a capturing pipeline onto a protocol or transport; the shell tests’ FakeTransport carries one so command tests can assert on tr.dispatchedEvents.

Future work: diagnostics to twisted.logger

The event side is complete; the diagnostic side is still on the legacy twisted.python.log API (~490 log.msg / log.err call sites). Now that no consumer parses the log stream, those can convert to per-class twisted.logger.Logger instances. The observer side is already there – python/logfile.py writes cowrie.log through textFileLogObserver and the twistd plugin registers ILogObserver providers – so only the emitters need converting. What the conversion buys:

Levels and namespaces

Per-class Logger instances give every line a namespace (cowrie.ssh.transport, cowrie.commands.wget) and a real level. Combined with LogLevelFilterPredicate this yields per-subsystem verbosity control from configuration – debug a single subsystem in production without drowning in the rest. Today the log has exactly one volume setting: everything.

Lazy formatting

self._log.debug("block {n} received", n=num) costs nearly nothing when filtered out, so debug statements can stay in the code permanently instead of being commented in and out during development.

Failures

self._log.failure("wget transfer failed") replaces log.err with explicit tracebacks attached as structured data.

A small number of session-less emitters (backend pool, proxy backend bookkeeping) also remain on log.msg(eventid=...); they never reached the output plugins historically, and making them first-class events is a separate follow-up.

Why not twisted.logger for events too

Twisted’s own answer to structured events is the twisted.logger package: Logger objects emitting keyword-structured events into observer chains. It is the right modernization for Cowrie’s diagnostic logging (above), but not for the event pipeline: twisted.logger has no bound context that survives a deferred hop, so events fired from a download callback would lose their attribution – exactly the failure the EventLog exists to prevent. A ContextVar holding the current session’s emitter was also considered and rejected: every reactor entry point would have to set it, and one missed entry point misattributes events silently. In a security event stream, a loud AttributeError on a missing self.events is preferable to quietly wrong data.

The chosen design – an explicit emitter object delivered through a single publisher – is not how Twisted routes its own logs, and that is deliberate: security events are domain data with delivery guarantees, not diagnostics. It does follow Twisted’s shape where it matters: EventDispatcher is structurally a domain-specific LogPublisher, including the per-observer error isolation Twisted’s publisher has.