cf7/ssh-honeyhill
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages 1 Wiki Activity Actions

changing default for log rotation

Browse source
This commit is contained in:
Caffeine Fueled 2026-08-06 21:39:52 +02:00
parent 1fd4b9f7c1
commit 45efdf1524
Signed by: cf7
GPG key ID: CA295D643074C68C
1 changed files with 55 additions and 22 deletions
Download patch file Download diff file Expand all files Collapse all files

77
ssh_honeypot.py
View file

@ -38,23 +38,43 @@ DEFAULT_BANNER = "SSH-2.0-OpenSSH_8.9p1 Ubuntu-3ubuntu0.4"
_event_logger = logging.getLogger("honeypot.events") # -> human-readable
_json_logger = logging.getLogger("honeypot.json") # -> JSON Lines
# Size at which a log file rotates — only applies when rotation is enabled
# (see _log_handler).
LOG_MAX_BYTES = 100 * 1024 * 1024
def setup_logging(log_dir: str) -> None:
def _log_handler(path: str, backup_count: int) -> RotatingFileHandler:
"""File handler for one log file; ``backup_count <= 0`` disables rotation.
Rotation only ever *prunes* — the oldest segment is deleted to make room.
For a honeypot that is the wrong default: losing captured attempts is worse
than a large file. So unless a backup count is configured we pass
``maxBytes=0``, which makes RotatingFileHandler never roll over, and the
file simply grows. Watch disk usage if you leave it that way.
"""
return RotatingFileHandler(
path,
maxBytes=LOG_MAX_BYTES if backup_count > 0 else 0,
backupCount=max(backup_count, 0),
encoding="utf-8",
)
def setup_logging(log_dir: str, backup_count: int = 0) -> None:
"""Configure three sinks: JSON Lines file, text file, and stdout.
Both ``_event_logger`` and ``_json_logger`` are configured here. They do
not propagate to the root logger so paramiko's transport chatter stays out
of our event logs.
``backup_count`` is how many rotated segments to keep per file; 0 (the
default) means never rotate. The console copy is never rotated — that is
the container runtime's job.
"""
os.makedirs(log_dir, exist_ok=True)
# JSON Lines: one json.dumps() record per line, no extra formatting.
json_handler = RotatingFileHandler(
os.path.join(log_dir, "honeypot.jsonl"),
maxBytes=10 * 1024 * 1024,
backupCount=5,
encoding="utf-8",
)
json_handler = _log_handler(os.path.join(log_dir, "honeypot.jsonl"), backup_count)
json_handler.setFormatter(logging.Formatter("%(message)s"))
_json_logger.addHandler(json_handler)
_json_logger.setLevel(logging.INFO)
@ -64,12 +84,7 @@ def setup_logging(log_dir: str) -> None:
# Local-time, space-separated, no timezone clutter — easy to scan in a terminal.
text_fmt = logging.Formatter("%(asctime)s %(message)s", "%Y-%m-%d %H:%M:%S")
text_handler = RotatingFileHandler(
os.path.join(log_dir, "honeypot.log"),
maxBytes=10 * 1024 * 1024,
backupCount=5,
encoding="utf-8",
)
text_handler = _log_handler(os.path.join(log_dir, "honeypot.log"), backup_count)
text_handler.setFormatter(text_fmt)
stdout_handler = logging.StreamHandler() # defaults to stderr/stdout stream
@ -80,8 +95,15 @@ def setup_logging(log_dir: str) -> None:
_event_logger.setLevel(logging.INFO)
_event_logger.propagate = False
# Keep paramiko's internal logging quiet but capture genuine errors.
logging.getLogger("paramiko").setLevel(logging.WARNING)
# Silence paramiko's own logging. An internet-facing honeypot constantly
# gets half-open / non-SSH probes (port scanners, nc, health checks) that
# make paramiko emit noisy "Error reading SSH protocol banner" tracebacks.
# We already record a clean `connection_error` line for these ourselves, so
# raise paramiko above ERROR and give it a NullHandler so nothing falls
# through to Python's last-resort stderr handler.
paramiko_log = logging.getLogger("paramiko")
paramiko_log.setLevel(logging.CRITICAL)
paramiko_log.addHandler(logging.NullHandler())
def _sanitize(value) -> str:
@ -105,7 +127,10 @@ def _human_message(event: str, fields: dict) -> str:
fields that matter) instead of a generic key=value dump, so a stream of
attempts is easy to read at a glance.
"""
src = f"{fields['src_ip']}:{fields.get('src_port', '?')}" if "src_ip" in fields else ""
# Source IP and port as two separate aligned columns (IPv4 fits in 15, port
# in 5). They are already separate fields in the JSON Lines output.
src = (f"ip={fields['src_ip']:<15} port={fields.get('src_port', '?')!s:<5}"
if "src_ip" in fields else "")
if event == "login_attempt":
if fields.get("auth_method") == "password":
@ -114,13 +139,13 @@ def _human_message(event: str, fields: dict) -> str:
cred = (f"user={fields.get('username')!r} "
f"key={fields.get('key_type')}/{fields.get('key_fingerprint')}")
client = _sanitize(fields.get("client_version", ""))
return f"{src:<21} {cred} [{fields.get('auth_method')}] {client}".rstrip()
return f"{src} {cred} [{fields.get('auth_method')}] {client}".rstrip()
if event == "connection":
return f"{src:<21} new connection"
return f"{src} new connection"
if event == "disconnect":
return f"{src:<21} disconnected"
return f"{src} disconnected"
if event == "connection_error":
return f"{src:<21} error: {_sanitize(fields.get('error'))}"
return f"{src} error: {_sanitize(fields.get('error'))}"
if event == "server_start":
return (f"listening on {fields.get('bind')}:{fields.get('port')} "
f"banner={fields.get('banner')!r} fingerprint={fields.get('host_key_fingerprint')}")
@ -345,7 +370,10 @@ class Honeypot:
# --------------------------------------------------------------------------- #
def _env_default(name: str, fallback: str) -> str:
return os.environ.get(name, fallback)
# `or fallback` rather than a .get() default: a set-but-empty variable is
# common in quadlet/compose files and systemd unit drop-ins, and .get()
# would hand back "" — which then crashes int() for the numeric options.
return os.environ.get(name) or fallback
def parse_args(argv=None) -> argparse.Namespace:
@ -359,6 +387,11 @@ def parse_args(argv=None) -> argparse.Namespace:
help="Path to the persisted host key (env HONEYPOT_HOST_KEY)")
p.add_argument("--log-dir", default=_env_default("HONEYPOT_LOG_DIR", "./logs"),
help="Directory for log files (env HONEYPOT_LOG_DIR)")
p.add_argument("--log-backups", type=int,
default=int(_env_default("HONEYPOT_LOG_BACKUPS", "0")),
help="Rotated log segments to keep per file, rotating every "
f"{LOG_MAX_BYTES // (1024 * 1024)} MB. 0 (the default) never "
"rotates and lets the files grow (env HONEYPOT_LOG_BACKUPS)")
p.add_argument("--banner", default=_env_default("HONEYPOT_BANNER", DEFAULT_BANNER),
help="Spoofed SSH version string (env HONEYPOT_BANNER)")
p.add_argument("--max-attempts", type=int,
@ -372,7 +405,7 @@ def parse_args(argv=None) -> argparse.Namespace:
def main(argv=None) -> int:
args = parse_args(argv)
setup_logging(args.log_dir)
setup_logging(args.log_dir, args.log_backups)
if not args.banner.startswith("SSH-2.0-"):
log_event("config_error", error="banner must start with 'SSH-2.0-'")