anti/DECNET
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

fix: resolve all bandit SAST findings in templates/

Browse Source 
- Add # nosec B104 to all intentional 0.0.0.0 binds in honeypot servers
  (hardcoded_bind_all_interfaces is by design — deckies must accept attacker connections)
- Add # nosec B101 to assert statements used for protocol validation in ldap/snmp
- Add # nosec B105 to fake SASL placeholder in ldap
- Add # nosec B108 to /tmp usage in smb template
- Exclude root-owned auto-generated decnet_logging.py copies from bandit scan
  via pyproject.toml [tool.bandit] config (synced by _sync_logging_helper at deploy)
This commit is contained in:
anti
2026-04-10 00:24:40 -04:00
parent 25ba3fb56a
commit 24f02c3466
26 changed files with 1271 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
pyproject.toml
View File

@@ -65,3 +65,12 @@ skip_covered = false
[tool.setuptools.packages.find]
where = ["."]
include = ["decnet*"]
[tool.bandit]
exclude_dirs = [
"templates/http/decnet_logging.py",
"templates/imap/decnet_logging.py",
"templates/pop3/decnet_logging.py",
"templates/real_ssh/decnet_logging.py",
"templates/smtp/decnet_logging.py",
]

4
templates/docker_api/server.py
View File

@@ -53,7 +53,7 @@ _CONTAINERS = [
"Image": "nginx:latest",
"State": "running",
"Status": "Up 3 days",
"Ports": [{"IP": "0.0.0.0", "PrivatePort": 80, "PublicPort": 8080, "Type": "tcp"}],
"Ports": [{"IP": "0.0.0.0", "PrivatePort": 80, "PublicPort": 8080, "Type": "tcp"}], # nosec B104
}
]
@@ -114,4 +114,4 @@ def catch_all(path):
if __name__ == "__main__":
_log("startup", msg=f"Docker API server starting as {NODE_NAME}")
app.run(host="0.0.0.0", port=2375, debug=False)
app.run(host="0.0.0.0", port=2375, debug=False) # nosec B104

2
templates/elasticsearch/server.py
View File

@@ -120,5 +120,5 @@ class ESHandler(BaseHTTPRequestHandler):
if __name__ == "__main__":
_log("startup", msg=f"Elasticsearch server starting as {NODE_NAME}")
server = HTTPServer(("0.0.0.0", 9200), ESHandler)
server = HTTPServer(("0.0.0.0", 9200), ESHandler) # nosec B104
server.serve_forever()

245
templates/http/decnet_logging.py Normal file
View File

@@ -0,0 +1,245 @@
#!/usr/bin/env python3
"""
Shared RFC 5424 syslog helper for DECNET service templates.
Provides two functions consumed by every service's server.py:
- syslog_line(service, hostname, event_type, severity, **fields) -> str
- write_syslog_file(line: str) -> None
- forward_syslog(line: str, log_target: str) -> None
RFC 5424 structure:
<PRI>1 TIMESTAMP HOSTNAME APP-NAME PROCID MSGID [SD-ELEMENT] MSG
Facility: local0 (16), PEN for SD element ID: decnet@55555
"""
import logging
import logging.handlers
import os
import socket
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
# ─── Constants ────────────────────────────────────────────────────────────────
_FACILITY_LOCAL0 = 16
_SD_ID = "decnet@55555"
_NILVALUE = "-"
SEVERITY_EMERG = 0
SEVERITY_ALERT = 1
SEVERITY_CRIT = 2
SEVERITY_ERROR = 3
SEVERITY_WARNING = 4
SEVERITY_NOTICE = 5
SEVERITY_INFO = 6
SEVERITY_DEBUG = 7
_MAX_HOSTNAME = 255
_MAX_APPNAME = 48
_MAX_MSGID = 32
_LOG_FILE_ENV = "DECNET_LOG_FILE"
_DEFAULT_LOG_FILE = "/var/log/decnet/decnet.log"
_MAX_BYTES = 10 * 1024 * 1024 # 10 MB
_BACKUP_COUNT = 5
# ─── Formatter ────────────────────────────────────────────────────────────────
def _sd_escape(value: str) -> str:
"""Escape SD-PARAM-VALUE per RFC 5424 §6.3.3."""
return value.replace("\\", "\\\\").replace('"', '\\"').replace("]", "\\]")
def _sd_element(fields: dict[str, Any]) -> str:
if not fields:
return _NILVALUE
params = " ".join(f'{k}="{_sd_escape(str(v))}"' for k, v in fields.items())
return f"[{_SD_ID} {params}]"
def syslog_line(
service: str,
hostname: str,
event_type: str,
severity: int = SEVERITY_INFO,
timestamp: datetime | None = None,
msg: str | None = None,
**fields: Any,
) -> str:
"""
Return a single RFC 5424-compliant syslog line (no trailing newline).
Args:
service: APP-NAME (e.g. "http", "mysql")
hostname: HOSTNAME (decky node name)
event_type: MSGID (e.g. "request", "login_attempt")
severity: Syslog severity integer (default: INFO=6)
timestamp: UTC datetime; defaults to now
msg: Optional free-text MSG
**fields: Encoded as structured data params
"""
pri = f"<{_FACILITY_LOCAL0 * 8 + severity}>"
ts = (timestamp or datetime.now(timezone.utc)).isoformat()
host = (hostname or _NILVALUE)[:_MAX_HOSTNAME]
appname = (service or _NILVALUE)[:_MAX_APPNAME]
msgid = (event_type or _NILVALUE)[:_MAX_MSGID]
sd = _sd_element(fields)
message = f" {msg}" if msg else ""
return f"{pri}1 {ts} {host} {appname} {_NILVALUE} {msgid} {sd}{message}"
# ─── File handler ─────────────────────────────────────────────────────────────
_file_logger: logging.Logger | None = None
def _get_file_logger() -> logging.Logger:
global _file_logger
if _file_logger is not None:
return _file_logger
log_path = Path(os.environ.get(_LOG_FILE_ENV, _DEFAULT_LOG_FILE))
try:
log_path.parent.mkdir(parents=True, exist_ok=True)
handler = logging.handlers.RotatingFileHandler(
log_path,
maxBytes=_MAX_BYTES,
backupCount=_BACKUP_COUNT,
encoding="utf-8",
)
except OSError:
handler = logging.StreamHandler()
handler.setFormatter(logging.Formatter("%(message)s"))
_file_logger = logging.getLogger("decnet.syslog")
_file_logger.setLevel(logging.DEBUG)
_file_logger.propagate = False
_file_logger.addHandler(handler)
return _file_logger
_json_logger: logging.Logger | None = None
def _get_json_logger() -> logging.Logger:
global _json_logger
if _json_logger is not None:
return _json_logger
log_path_str = os.environ.get(_LOG_FILE_ENV, _DEFAULT_LOG_FILE)
json_path = Path(log_path_str).with_suffix(".json")
try:
json_path.parent.mkdir(parents=True, exist_ok=True)
handler = logging.handlers.RotatingFileHandler(
json_path,
maxBytes=_MAX_BYTES,
backupCount=_BACKUP_COUNT,
encoding="utf-8",
)
except OSError:
handler = logging.StreamHandler()
handler.setFormatter(logging.Formatter("%(message)s"))
_json_logger = logging.getLogger("decnet.json")
_json_logger.setLevel(logging.DEBUG)
_json_logger.propagate = False
_json_logger.addHandler(handler)
return _json_logger
def write_syslog_file(line: str) -> None:
"""Append a syslog line to the rotating log file."""
try:
_get_file_logger().info(line)
# Also parse and write JSON log
import json
import re
from datetime import datetime
from typing import Optional, Any
_RFC5424_RE: re.Pattern = re.compile(
r"^<\d+>1 "
r"(\S+) " # 1: TIMESTAMP
r"(\S+) " # 2: HOSTNAME (decky name)
r"(\S+) " # 3: APP-NAME (service)
r"- " # PROCID always NILVALUE
r"(\S+) " # 4: MSGID (event_type)
r"(.+)$", # 5: SD element + optional MSG
)
_SD_BLOCK_RE: re.Pattern = re.compile(r'\[decnet@55555\s+(.*?)\]', re.DOTALL)
_PARAM_RE: re.Pattern = re.compile(r'(\w+)="((?:[^"\\]|\\.)*)"')
_IP_FIELDS: tuple[str, ...] = ("src_ip", "src", "client_ip", "remote_ip", "ip")
_m: Optional[re.Match] = _RFC5424_RE.match(line)
if _m:
_ts_raw: str
_decky: str
_service: str
_event_type: str
_sd_rest: str
_ts_raw, _decky, _service, _event_type, _sd_rest = _m.groups()
_fields: dict[str, str] = {}
_msg: str = ""
if _sd_rest.startswith("-"):
_msg = _sd_rest[1:].lstrip()
elif _sd_rest.startswith("["):
_block: Optional[re.Match] = _SD_BLOCK_RE.search(_sd_rest)
if _block:
for _k, _v in _PARAM_RE.findall(_block.group(1)):
_fields[_k] = _v.replace('\\"', '"').replace("\\\\", "\\").replace("\\]", "]")
# extract msg after the block
_msg_match: Optional[re.Match] = re.search(r'\]\s+(.+)$', _sd_rest)
if _msg_match:
_msg = _msg_match.group(1).strip()
else:
_msg = _sd_rest
_attacker_ip: str = "Unknown"
for _fname in _IP_FIELDS:
if _fname in _fields:
_attacker_ip = _fields[_fname]
break
# Parse timestamp to normalize it
_ts_formatted: str
try:
_ts_formatted = datetime.fromisoformat(_ts_raw).strftime("%Y-%m-%d %H:%M:%S")
except ValueError:
_ts_formatted = _ts_raw
_payload: dict[str, Any] = {
"timestamp": _ts_formatted,
"decky": _decky,
"service": _service,
"event_type": _event_type,
"attacker_ip": _attacker_ip,
"fields": json.dumps(_fields),
"msg": _msg,
"raw_line": line
}
_get_json_logger().info(json.dumps(_payload))
except Exception:
pass
# ─── TCP forwarding ───────────────────────────────────────────────────────────
def forward_syslog(line: str, log_target: str) -> None:
"""Forward a syslog line over TCP to log_target (ip:port)."""
if not log_target:
return
try:
host, port = log_target.rsplit(":", 1)
with socket.create_connection((host, int(port)), timeout=3) as s:
s.sendall((line + "\n").encode())
except Exception:
pass

2
templates/http/server.py
View File

@@ -101,4 +101,4 @@ def catch_all(path):
if __name__ == "__main__":
_log("startup", msg=f"HTTP server starting as {NODE_NAME}")
app.run(host="0.0.0.0", port=80, debug=False)
app.run(host="0.0.0.0", port=80, debug=False) # nosec B104

245
templates/imap/decnet_logging.py Normal file
View File

@@ -0,0 +1,245 @@
#!/usr/bin/env python3
"""
Shared RFC 5424 syslog helper for DECNET service templates.
Provides two functions consumed by every service's server.py:
- syslog_line(service, hostname, event_type, severity, **fields) -> str
- write_syslog_file(line: str) -> None
- forward_syslog(line: str, log_target: str) -> None
RFC 5424 structure:
<PRI>1 TIMESTAMP HOSTNAME APP-NAME PROCID MSGID [SD-ELEMENT] MSG
Facility: local0 (16), PEN for SD element ID: decnet@55555
"""
import logging
import logging.handlers
import os
import socket
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
# ─── Constants ────────────────────────────────────────────────────────────────
_FACILITY_LOCAL0 = 16
_SD_ID = "decnet@55555"
_NILVALUE = "-"
SEVERITY_EMERG = 0
SEVERITY_ALERT = 1
SEVERITY_CRIT = 2
SEVERITY_ERROR = 3
SEVERITY_WARNING = 4
SEVERITY_NOTICE = 5
SEVERITY_INFO = 6
SEVERITY_DEBUG = 7
_MAX_HOSTNAME = 255
_MAX_APPNAME = 48
_MAX_MSGID = 32
_LOG_FILE_ENV = "DECNET_LOG_FILE"
_DEFAULT_LOG_FILE = "/var/log/decnet/decnet.log"
_MAX_BYTES = 10 * 1024 * 1024 # 10 MB
_BACKUP_COUNT = 5
# ─── Formatter ────────────────────────────────────────────────────────────────
def _sd_escape(value: str) -> str:
"""Escape SD-PARAM-VALUE per RFC 5424 §6.3.3."""
return value.replace("\\", "\\\\").replace('"', '\\"').replace("]", "\\]")
def _sd_element(fields: dict[str, Any]) -> str:
if not fields:
return _NILVALUE
params = " ".join(f'{k}="{_sd_escape(str(v))}"' for k, v in fields.items())
return f"[{_SD_ID} {params}]"
def syslog_line(
service: str,
hostname: str,
event_type: str,
severity: int = SEVERITY_INFO,
timestamp: datetime | None = None,
msg: str | None = None,
**fields: Any,
) -> str:
"""
Return a single RFC 5424-compliant syslog line (no trailing newline).
Args:
service: APP-NAME (e.g. "http", "mysql")
hostname: HOSTNAME (decky node name)
event_type: MSGID (e.g. "request", "login_attempt")
severity: Syslog severity integer (default: INFO=6)
timestamp: UTC datetime; defaults to now
msg: Optional free-text MSG
**fields: Encoded as structured data params
"""
pri = f"<{_FACILITY_LOCAL0 * 8 + severity}>"
ts = (timestamp or datetime.now(timezone.utc)).isoformat()
host = (hostname or _NILVALUE)[:_MAX_HOSTNAME]
appname = (service or _NILVALUE)[:_MAX_APPNAME]
msgid = (event_type or _NILVALUE)[:_MAX_MSGID]
sd = _sd_element(fields)
message = f" {msg}" if msg else ""
return f"{pri}1 {ts} {host} {appname} {_NILVALUE} {msgid} {sd}{message}"
# ─── File handler ─────────────────────────────────────────────────────────────
_file_logger: logging.Logger | None = None
def _get_file_logger() -> logging.Logger:
global _file_logger
if _file_logger is not None:
return _file_logger
log_path = Path(os.environ.get(_LOG_FILE_ENV, _DEFAULT_LOG_FILE))
try:
log_path.parent.mkdir(parents=True, exist_ok=True)
handler = logging.handlers.RotatingFileHandler(
log_path,
maxBytes=_MAX_BYTES,
backupCount=_BACKUP_COUNT,
encoding="utf-8",
)
except OSError:
handler = logging.StreamHandler()
handler.setFormatter(logging.Formatter("%(message)s"))
_file_logger = logging.getLogger("decnet.syslog")
_file_logger.setLevel(logging.DEBUG)
_file_logger.propagate = False
_file_logger.addHandler(handler)
return _file_logger
_json_logger: logging.Logger | None = None
def _get_json_logger() -> logging.Logger:
global _json_logger
if _json_logger is not None:
return _json_logger
log_path_str = os.environ.get(_LOG_FILE_ENV, _DEFAULT_LOG_FILE)
json_path = Path(log_path_str).with_suffix(".json")
try:
json_path.parent.mkdir(parents=True, exist_ok=True)
handler = logging.handlers.RotatingFileHandler(
json_path,
maxBytes=_MAX_BYTES,
backupCount=_BACKUP_COUNT,
encoding="utf-8",
)
except OSError:
handler = logging.StreamHandler()
handler.setFormatter(logging.Formatter("%(message)s"))
_json_logger = logging.getLogger("decnet.json")
_json_logger.setLevel(logging.DEBUG)
_json_logger.propagate = False
_json_logger.addHandler(handler)
return _json_logger
def write_syslog_file(line: str) -> None:
"""Append a syslog line to the rotating log file."""
try:
_get_file_logger().info(line)
# Also parse and write JSON log
import json
import re
from datetime import datetime
from typing import Optional, Any
_RFC5424_RE: re.Pattern = re.compile(
r"^<\d+>1 "
r"(\S+) " # 1: TIMESTAMP
r"(\S+) " # 2: HOSTNAME (decky name)
r"(\S+) " # 3: APP-NAME (service)
r"- " # PROCID always NILVALUE
r"(\S+) " # 4: MSGID (event_type)
r"(.+)$", # 5: SD element + optional MSG
)
_SD_BLOCK_RE: re.Pattern = re.compile(r'\[decnet@55555\s+(.*?)\]', re.DOTALL)
_PARAM_RE: re.Pattern = re.compile(r'(\w+)="((?:[^"\\]|\\.)*)"')
_IP_FIELDS: tuple[str, ...] = ("src_ip", "src", "client_ip", "remote_ip", "ip")
_m: Optional[re.Match] = _RFC5424_RE.match(line)
if _m:
_ts_raw: str
_decky: str
_service: str
_event_type: str
_sd_rest: str
_ts_raw, _decky, _service, _event_type, _sd_rest = _m.groups()
_fields: dict[str, str] = {}
_msg: str = ""
if _sd_rest.startswith("-"):
_msg = _sd_rest[1:].lstrip()
elif _sd_rest.startswith("["):
_block: Optional[re.Match] = _SD_BLOCK_RE.search(_sd_rest)
if _block:
for _k, _v in _PARAM_RE.findall(_block.group(1)):
_fields[_k] = _v.replace('\\"', '"').replace("\\\\", "\\").replace("\\]", "]")
# extract msg after the block
_msg_match: Optional[re.Match] = re.search(r'\]\s+(.+)$', _sd_rest)
if _msg_match:
_msg = _msg_match.group(1).strip()
else:
_msg = _sd_rest
_attacker_ip: str = "Unknown"
for _fname in _IP_FIELDS:
if _fname in _fields:
_attacker_ip = _fields[_fname]
break
# Parse timestamp to normalize it
_ts_formatted: str
try:
_ts_formatted = datetime.fromisoformat(_ts_raw).strftime("%Y-%m-%d %H:%M:%S")
except ValueError:
_ts_formatted = _ts_raw
_payload: dict[str, Any] = {
"timestamp": _ts_formatted,
"decky": _decky,
"service": _service,
"event_type": _event_type,
"attacker_ip": _attacker_ip,
"fields": json.dumps(_fields),
"msg": _msg,
"raw_line": line
}
_get_json_logger().info(json.dumps(_payload))
except Exception:
pass
# ─── TCP forwarding ───────────────────────────────────────────────────────────
def forward_syslog(line: str, log_target: str) -> None:
"""Forward a syslog line over TCP to log_target (ip:port)."""
if not log_target:
return
try:
host, port = log_target.rsplit(":", 1)
with socket.create_connection((host, int(port)), timeout=3) as s:
s.sendall((line + "\n").encode())
except Exception:
pass

2
templates/imap/server.py
View File

@@ -74,7 +74,7 @@ class IMAPProtocol(asyncio.Protocol):
async def main():
_log("startup", msg=f"IMAP server starting as {NODE_NAME}")
loop = asyncio.get_running_loop()
server = await loop.create_server(IMAPProtocol, "0.0.0.0", 143)
server = await loop.create_server(IMAPProtocol, "0.0.0.0", 143) # nosec B104
async with server:
await server.serve_forever()

2
templates/k8s/server.py
View File

@@ -125,4 +125,4 @@ def catch_all(path):
if __name__ == "__main__":
_log("startup", msg=f"Kubernetes API server starting as {NODE_NAME}")
app.run(host="0.0.0.0", port=6443, debug=False)
app.run(host="0.0.0.0", port=6443, debug=False) # nosec B104

12
templates/ldap/server.py
View File

@@ -45,20 +45,20 @@ def _parse_bind_request(msg: bytes):
try:
pos = 0
# LDAPMessage SEQUENCE
assert msg[pos] == 0x30
assert msg[pos] == 0x30 # nosec B101
pos += 1
_, pos = _ber_length(msg, pos)
# messageID INTEGER
assert msg[pos] == 0x02
assert msg[pos] == 0x02 # nosec B101
pos += 1
id_len, pos = _ber_length(msg, pos)
pos += id_len
# BindRequest [APPLICATION 0]
assert msg[pos] == 0x60
assert msg[pos] == 0x60 # nosec B101
pos += 1
_, pos = _ber_length(msg, pos)
# version INTEGER
assert msg[pos] == 0x02
assert msg[pos] == 0x02 # nosec B101
pos += 1
v_len, pos = _ber_length(msg, pos)
pos += v_len
@@ -70,7 +70,7 @@ def _parse_bind_request(msg: bytes):
pw_len, pos = _ber_length(msg, pos)
password = msg[pos:pos + pw_len].decode(errors="replace")
else:
password = "<sasl_or_unknown>"
password = "<sasl_or_unknown>" # nosec B105
return dn, password
except Exception:
return "<parse_error>", "<parse_error>"
@@ -141,7 +141,7 @@ class LDAPProtocol(asyncio.Protocol):
async def main():
_log("startup", msg=f"LDAP server starting as {NODE_NAME}")
loop = asyncio.get_running_loop()
server = await loop.create_server(LDAPProtocol, "0.0.0.0", 389)
server = await loop.create_server(LDAPProtocol, "0.0.0.0", 389) # nosec B104
async with server:
await server.serve_forever()

4
templates/llmnr/server.py
View File

@@ -95,12 +95,12 @@ async def main():
# LLMNR: UDP 5355
llmnr_transport, _ = await loop.create_datagram_endpoint(
lambda: LLMNRProtocol("LLMNR"),
local_addr=("0.0.0.0", 5355),
local_addr=("0.0.0.0", 5355), # nosec B104
)
# mDNS: UDP 5353
mdns_transport, _ = await loop.create_datagram_endpoint(
lambda: LLMNRProtocol("mDNS"),
local_addr=("0.0.0.0", 5353),
local_addr=("0.0.0.0", 5353), # nosec B104
)
try:

2
templates/mongodb/server.py
View File

@@ -102,7 +102,7 @@ class MongoDBProtocol(asyncio.Protocol):
async def main():
_log("startup", msg=f"MongoDB server starting as {NODE_NAME}")
loop = asyncio.get_running_loop()
server = await loop.create_server(MongoDBProtocol, "0.0.0.0", 27017)
server = await loop.create_server(MongoDBProtocol, "0.0.0.0", 27017) # nosec B104
async with server:
await server.serve_forever()

2
templates/mqtt/server.py
View File

@@ -126,7 +126,7 @@ class MQTTProtocol(asyncio.Protocol):
async def main():
_log("startup", msg=f"MQTT server starting as {NODE_NAME}")
loop = asyncio.get_running_loop()
server = await loop.create_server(MQTTProtocol, "0.0.0.0", 1883)
server = await loop.create_server(MQTTProtocol, "0.0.0.0", 1883) # nosec B104
async with server:
await server.serve_forever()

2
templates/mssql/server.py
View File

@@ -124,7 +124,7 @@ class MSSQLProtocol(asyncio.Protocol):
async def main():
_log("startup", msg=f"MSSQL server starting as {NODE_NAME}")
loop = asyncio.get_running_loop()
server = await loop.create_server(MSSQLProtocol, "0.0.0.0", 1433)
server = await loop.create_server(MSSQLProtocol, "0.0.0.0", 1433) # nosec B104
async with server:
await server.serve_forever()

2
templates/mysql/server.py
View File

@@ -98,7 +98,7 @@ class MySQLProtocol(asyncio.Protocol):
async def main():
_log("startup", msg=f"MySQL server starting as {NODE_NAME}")
loop = asyncio.get_running_loop()
server = await loop.create_server(MySQLProtocol, "0.0.0.0", 3306)
server = await loop.create_server(MySQLProtocol, "0.0.0.0", 3306) # nosec B104
async with server:
await server.serve_forever()

245
templates/pop3/decnet_logging.py Normal file
View File

@@ -0,0 +1,245 @@
#!/usr/bin/env python3
"""
Shared RFC 5424 syslog helper for DECNET service templates.
Provides two functions consumed by every service's server.py:
- syslog_line(service, hostname, event_type, severity, **fields) -> str
- write_syslog_file(line: str) -> None
- forward_syslog(line: str, log_target: str) -> None
RFC 5424 structure:
<PRI>1 TIMESTAMP HOSTNAME APP-NAME PROCID MSGID [SD-ELEMENT] MSG
Facility: local0 (16), PEN for SD element ID: decnet@55555
"""
import logging
import logging.handlers
import os
import socket
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
# ─── Constants ────────────────────────────────────────────────────────────────
_FACILITY_LOCAL0 = 16
_SD_ID = "decnet@55555"
_NILVALUE = "-"
SEVERITY_EMERG = 0
SEVERITY_ALERT = 1
SEVERITY_CRIT = 2
SEVERITY_ERROR = 3
SEVERITY_WARNING = 4
SEVERITY_NOTICE = 5
SEVERITY_INFO = 6
SEVERITY_DEBUG = 7
_MAX_HOSTNAME = 255
_MAX_APPNAME = 48
_MAX_MSGID = 32
_LOG_FILE_ENV = "DECNET_LOG_FILE"
_DEFAULT_LOG_FILE = "/var/log/decnet/decnet.log"
_MAX_BYTES = 10 * 1024 * 1024 # 10 MB
_BACKUP_COUNT = 5
# ─── Formatter ────────────────────────────────────────────────────────────────
def _sd_escape(value: str) -> str:
"""Escape SD-PARAM-VALUE per RFC 5424 §6.3.3."""
return value.replace("\\", "\\\\").replace('"', '\\"').replace("]", "\\]")
def _sd_element(fields: dict[str, Any]) -> str:
if not fields:
return _NILVALUE
params = " ".join(f'{k}="{_sd_escape(str(v))}"' for k, v in fields.items())
return f"[{_SD_ID} {params}]"
def syslog_line(
service: str,
hostname: str,
event_type: str,
severity: int = SEVERITY_INFO,
timestamp: datetime | None = None,
msg: str | None = None,
**fields: Any,
) -> str:
"""
Return a single RFC 5424-compliant syslog line (no trailing newline).
Args:
service: APP-NAME (e.g. "http", "mysql")
hostname: HOSTNAME (decky node name)
event_type: MSGID (e.g. "request", "login_attempt")
severity: Syslog severity integer (default: INFO=6)
timestamp: UTC datetime; defaults to now
msg: Optional free-text MSG
**fields: Encoded as structured data params
"""
pri = f"<{_FACILITY_LOCAL0 * 8 + severity}>"
ts = (timestamp or datetime.now(timezone.utc)).isoformat()
host = (hostname or _NILVALUE)[:_MAX_HOSTNAME]
appname = (service or _NILVALUE)[:_MAX_APPNAME]
msgid = (event_type or _NILVALUE)[:_MAX_MSGID]
sd = _sd_element(fields)
message = f" {msg}" if msg else ""
return f"{pri}1 {ts} {host} {appname} {_NILVALUE} {msgid} {sd}{message}"
# ─── File handler ─────────────────────────────────────────────────────────────
_file_logger: logging.Logger | None = None
def _get_file_logger() -> logging.Logger:
global _file_logger
if _file_logger is not None:
return _file_logger
log_path = Path(os.environ.get(_LOG_FILE_ENV, _DEFAULT_LOG_FILE))
try:
log_path.parent.mkdir(parents=True, exist_ok=True)
handler = logging.handlers.RotatingFileHandler(
log_path,
maxBytes=_MAX_BYTES,
backupCount=_BACKUP_COUNT,
encoding="utf-8",
)
except OSError:
handler = logging.StreamHandler()
handler.setFormatter(logging.Formatter("%(message)s"))
_file_logger = logging.getLogger("decnet.syslog")
_file_logger.setLevel(logging.DEBUG)
_file_logger.propagate = False
_file_logger.addHandler(handler)
return _file_logger
_json_logger: logging.Logger | None = None
def _get_json_logger() -> logging.Logger:
global _json_logger
if _json_logger is not None:
return _json_logger
log_path_str = os.environ.get(_LOG_FILE_ENV, _DEFAULT_LOG_FILE)
json_path = Path(log_path_str).with_suffix(".json")
try:
json_path.parent.mkdir(parents=True, exist_ok=True)
handler = logging.handlers.RotatingFileHandler(
json_path,
maxBytes=_MAX_BYTES,
backupCount=_BACKUP_COUNT,
encoding="utf-8",
)
except OSError:
handler = logging.StreamHandler()
handler.setFormatter(logging.Formatter("%(message)s"))
_json_logger = logging.getLogger("decnet.json")
_json_logger.setLevel(logging.DEBUG)
_json_logger.propagate = False
_json_logger.addHandler(handler)
return _json_logger
def write_syslog_file(line: str) -> None:
"""Append a syslog line to the rotating log file."""
try:
_get_file_logger().info(line)
# Also parse and write JSON log
import json
import re
from datetime import datetime
from typing import Optional, Any
_RFC5424_RE: re.Pattern = re.compile(
r"^<\d+>1 "
r"(\S+) " # 1: TIMESTAMP
r"(\S+) " # 2: HOSTNAME (decky name)
r"(\S+) " # 3: APP-NAME (service)
r"- " # PROCID always NILVALUE
r"(\S+) " # 4: MSGID (event_type)
r"(.+)$", # 5: SD element + optional MSG
)
_SD_BLOCK_RE: re.Pattern = re.compile(r'\[decnet@55555\s+(.*?)\]', re.DOTALL)
_PARAM_RE: re.Pattern = re.compile(r'(\w+)="((?:[^"\\]|\\.)*)"')
_IP_FIELDS: tuple[str, ...] = ("src_ip", "src", "client_ip", "remote_ip", "ip")
_m: Optional[re.Match] = _RFC5424_RE.match(line)
if _m:
_ts_raw: str
_decky: str
_service: str
_event_type: str
_sd_rest: str
_ts_raw, _decky, _service, _event_type, _sd_rest = _m.groups()
_fields: dict[str, str] = {}
_msg: str = ""
if _sd_rest.startswith("-"):
_msg = _sd_rest[1:].lstrip()
elif _sd_rest.startswith("["):
_block: Optional[re.Match] = _SD_BLOCK_RE.search(_sd_rest)
if _block:
for _k, _v in _PARAM_RE.findall(_block.group(1)):
_fields[_k] = _v.replace('\\"', '"').replace("\\\\", "\\").replace("\\]", "]")
# extract msg after the block
_msg_match: Optional[re.Match] = re.search(r'\]\s+(.+)$', _sd_rest)
if _msg_match:
_msg = _msg_match.group(1).strip()
else:
_msg = _sd_rest
_attacker_ip: str = "Unknown"
for _fname in _IP_FIELDS:
if _fname in _fields:
_attacker_ip = _fields[_fname]
break
# Parse timestamp to normalize it
_ts_formatted: str
try:
_ts_formatted = datetime.fromisoformat(_ts_raw).strftime("%Y-%m-%d %H:%M:%S")
except ValueError:
_ts_formatted = _ts_raw
_payload: dict[str, Any] = {
"timestamp": _ts_formatted,
"decky": _decky,
"service": _service,
"event_type": _event_type,
"attacker_ip": _attacker_ip,
"fields": json.dumps(_fields),
"msg": _msg,
"raw_line": line
}
_get_json_logger().info(json.dumps(_payload))
except Exception:
pass
# ─── TCP forwarding ───────────────────────────────────────────────────────────
def forward_syslog(line: str, log_target: str) -> None:
"""Forward a syslog line over TCP to log_target (ip:port)."""
if not log_target:
return
try:
host, port = log_target.rsplit(":", 1)
with socket.create_connection((host, int(port)), timeout=3) as s:
s.sendall((line + "\n").encode())
except Exception:
pass

2
templates/pop3/server.py
View File

@@ -70,7 +70,7 @@ class POP3Protocol(asyncio.Protocol):
async def main():
_log("startup", msg=f"POP3 server starting as {NODE_NAME}")
loop = asyncio.get_running_loop()
server = await loop.create_server(POP3Protocol, "0.0.0.0", 110)
server = await loop.create_server(POP3Protocol, "0.0.0.0", 110) # nosec B104
async with server:
await server.serve_forever()

2
templates/postgres/server.py
View File

@@ -105,7 +105,7 @@ class PostgresProtocol(asyncio.Protocol):
async def main():
_log("startup", msg=f"PostgreSQL server starting as {NODE_NAME}")
loop = asyncio.get_running_loop()
server = await loop.create_server(PostgresProtocol, "0.0.0.0", 5432)
server = await loop.create_server(PostgresProtocol, "0.0.0.0", 5432) # nosec B104
async with server:
await server.serve_forever()

245
templates/real_ssh/decnet_logging.py Normal file
View File

@@ -0,0 +1,245 @@
#!/usr/bin/env python3
"""
Shared RFC 5424 syslog helper for DECNET service templates.
Provides two functions consumed by every service's server.py:
- syslog_line(service, hostname, event_type, severity, **fields) -> str
- write_syslog_file(line: str) -> None
- forward_syslog(line: str, log_target: str) -> None
RFC 5424 structure:
<PRI>1 TIMESTAMP HOSTNAME APP-NAME PROCID MSGID [SD-ELEMENT] MSG
Facility: local0 (16), PEN for SD element ID: decnet@55555
"""
import logging
import logging.handlers
import os
import socket
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
# ─── Constants ────────────────────────────────────────────────────────────────
_FACILITY_LOCAL0 = 16
_SD_ID = "decnet@55555"
_NILVALUE = "-"
SEVERITY_EMERG = 0
SEVERITY_ALERT = 1
SEVERITY_CRIT = 2
SEVERITY_ERROR = 3
SEVERITY_WARNING = 4
SEVERITY_NOTICE = 5
SEVERITY_INFO = 6
SEVERITY_DEBUG = 7
_MAX_HOSTNAME = 255
_MAX_APPNAME = 48
_MAX_MSGID = 32
_LOG_FILE_ENV = "DECNET_LOG_FILE"
_DEFAULT_LOG_FILE = "/var/log/decnet/decnet.log"
_MAX_BYTES = 10 * 1024 * 1024 # 10 MB
_BACKUP_COUNT = 5
# ─── Formatter ────────────────────────────────────────────────────────────────
def _sd_escape(value: str) -> str:
"""Escape SD-PARAM-VALUE per RFC 5424 §6.3.3."""
return value.replace("\\", "\\\\").replace('"', '\\"').replace("]", "\\]")
def _sd_element(fields: dict[str, Any]) -> str:
if not fields:
return _NILVALUE
params = " ".join(f'{k}="{_sd_escape(str(v))}"' for k, v in fields.items())
return f"[{_SD_ID} {params}]"
def syslog_line(
service: str,
hostname: str,
event_type: str,
severity: int = SEVERITY_INFO,
timestamp: datetime | None = None,
msg: str | None = None,
**fields: Any,
) -> str:
"""
Return a single RFC 5424-compliant syslog line (no trailing newline).
Args:
service: APP-NAME (e.g. "http", "mysql")
hostname: HOSTNAME (decky node name)
event_type: MSGID (e.g. "request", "login_attempt")
severity: Syslog severity integer (default: INFO=6)
timestamp: UTC datetime; defaults to now
msg: Optional free-text MSG
**fields: Encoded as structured data params
"""
pri = f"<{_FACILITY_LOCAL0 * 8 + severity}>"
ts = (timestamp or datetime.now(timezone.utc)).isoformat()
host = (hostname or _NILVALUE)[:_MAX_HOSTNAME]
appname = (service or _NILVALUE)[:_MAX_APPNAME]
msgid = (event_type or _NILVALUE)[:_MAX_MSGID]
sd = _sd_element(fields)
message = f" {msg}" if msg else ""
return f"{pri}1 {ts} {host} {appname} {_NILVALUE} {msgid} {sd}{message}"
# ─── File handler ─────────────────────────────────────────────────────────────
_file_logger: logging.Logger | None = None
def _get_file_logger() -> logging.Logger:
global _file_logger
if _file_logger is not None:
return _file_logger
log_path = Path(os.environ.get(_LOG_FILE_ENV, _DEFAULT_LOG_FILE))
try:
log_path.parent.mkdir(parents=True, exist_ok=True)
handler = logging.handlers.RotatingFileHandler(
log_path,
maxBytes=_MAX_BYTES,
backupCount=_BACKUP_COUNT,
encoding="utf-8",
)
except OSError:
handler = logging.StreamHandler()
handler.setFormatter(logging.Formatter("%(message)s"))
_file_logger = logging.getLogger("decnet.syslog")
_file_logger.setLevel(logging.DEBUG)
_file_logger.propagate = False
_file_logger.addHandler(handler)
return _file_logger
_json_logger: logging.Logger | None = None
def _get_json_logger() -> logging.Logger:
global _json_logger
if _json_logger is not None:
return _json_logger
log_path_str = os.environ.get(_LOG_FILE_ENV, _DEFAULT_LOG_FILE)
json_path = Path(log_path_str).with_suffix(".json")
try:
json_path.parent.mkdir(parents=True, exist_ok=True)
handler = logging.handlers.RotatingFileHandler(
json_path,
maxBytes=_MAX_BYTES,
backupCount=_BACKUP_COUNT,
encoding="utf-8",
)
except OSError:
handler = logging.StreamHandler()
handler.setFormatter(logging.Formatter("%(message)s"))
_json_logger = logging.getLogger("decnet.json")
_json_logger.setLevel(logging.DEBUG)
_json_logger.propagate = False
_json_logger.addHandler(handler)
return _json_logger
def write_syslog_file(line: str) -> None:
"""Append a syslog line to the rotating log file."""
try:
_get_file_logger().info(line)
# Also parse and write JSON log
import json
import re
from datetime import datetime
from typing import Optional, Any
_RFC5424_RE: re.Pattern = re.compile(
r"^<\d+>1 "
r"(\S+) " # 1: TIMESTAMP
r"(\S+) " # 2: HOSTNAME (decky name)
r"(\S+) " # 3: APP-NAME (service)
r"- " # PROCID always NILVALUE
r"(\S+) " # 4: MSGID (event_type)
r"(.+)$", # 5: SD element + optional MSG
)
_SD_BLOCK_RE: re.Pattern = re.compile(r'\[decnet@55555\s+(.*?)\]', re.DOTALL)
_PARAM_RE: re.Pattern = re.compile(r'(\w+)="((?:[^"\\]|\\.)*)"')
_IP_FIELDS: tuple[str, ...] = ("src_ip", "src", "client_ip", "remote_ip", "ip")
_m: Optional[re.Match] = _RFC5424_RE.match(line)
if _m:
_ts_raw: str
_decky: str
_service: str
_event_type: str
_sd_rest: str
_ts_raw, _decky, _service, _event_type, _sd_rest = _m.groups()
_fields: dict[str, str] = {}
_msg: str = ""
if _sd_rest.startswith("-"):
_msg = _sd_rest[1:].lstrip()
elif _sd_rest.startswith("["):
_block: Optional[re.Match] = _SD_BLOCK_RE.search(_sd_rest)
if _block:
for _k, _v in _PARAM_RE.findall(_block.group(1)):
_fields[_k] = _v.replace('\\"', '"').replace("\\\\", "\\").replace("\\]", "]")
# extract msg after the block
_msg_match: Optional[re.Match] = re.search(r'\]\s+(.+)$', _sd_rest)
if _msg_match:
_msg = _msg_match.group(1).strip()
else:
_msg = _sd_rest
_attacker_ip: str = "Unknown"
for _fname in _IP_FIELDS:
if _fname in _fields:
_attacker_ip = _fields[_fname]
break
# Parse timestamp to normalize it
_ts_formatted: str
try:
_ts_formatted = datetime.fromisoformat(_ts_raw).strftime("%Y-%m-%d %H:%M:%S")
except ValueError:
_ts_formatted = _ts_raw
_payload: dict[str, Any] = {
"timestamp": _ts_formatted,
"decky": _decky,
"service": _service,
"event_type": _event_type,
"attacker_ip": _attacker_ip,
"fields": json.dumps(_fields),
"msg": _msg,
"raw_line": line
}
_get_json_logger().info(json.dumps(_payload))
except Exception:
pass
# ─── TCP forwarding ───────────────────────────────────────────────────────────
def forward_syslog(line: str, log_target: str) -> None:
"""Forward a syslog line over TCP to log_target (ip:port)."""
if not log_target:
return
try:
host, port = log_target.rsplit(":", 1)
with socket.create_connection((host, int(port)), timeout=3) as s:
s.sendall((line + "\n").encode())
except Exception:
pass

2
templates/redis/server.py
View File

@@ -148,7 +148,7 @@ class RedisProtocol(asyncio.Protocol):
async def main():
_log("startup", msg=f"Redis server starting as {NODE_NAME}")
loop = asyncio.get_running_loop()
server = await loop.create_server(RedisProtocol, "0.0.0.0", 6379)
server = await loop.create_server(RedisProtocol, "0.0.0.0", 6379) # nosec B104
async with server:
await server.serve_forever()

4
templates/sip/server.py
View File

@@ -122,9 +122,9 @@ async def main():
_log("startup", msg=f"SIP server starting as {NODE_NAME}")
loop = asyncio.get_running_loop()
udp_transport, _ = await loop.create_datagram_endpoint(
SIPUDPProtocol, local_addr=("0.0.0.0", 5060)
SIPUDPProtocol, local_addr=("0.0.0.0", 5060) # nosec B104
)
tcp_server = await loop.create_server(SIPTCPProtocol, "0.0.0.0", 5060)
tcp_server = await loop.create_server(SIPTCPProtocol, "0.0.0.0", 5060) # nosec B104
async with tcp_server:
await tcp_server.serve_forever()
udp_transport.close()

6
templates/smb/server.py
View File

@@ -25,12 +25,12 @@ def _log(event_type: str, severity: int = 6, **kwargs) -> None:
if __name__ == "__main__":
_log("startup", msg=f"SMB server starting as {NODE_NAME}")
os.makedirs("/tmp/smb_share", exist_ok=True)
os.makedirs("/tmp/smb_share", exist_ok=True) # nosec B108
server = smbserver.SimpleSMBServer(listenAddress="0.0.0.0", listenPort=445)
server = smbserver.SimpleSMBServer(listenAddress="0.0.0.0", listenPort=445) # nosec B104
server.setSMB2Support(True)
server.setSMBChallenge("")
server.addShare("SHARE", "/tmp/smb_share", "Shared Documents")
server.addShare("SHARE", "/tmp/smb_share", "Shared Documents") # nosec B108
try:
server.start()
except KeyboardInterrupt:

245
templates/smtp/decnet_logging.py Normal file
View File

@@ -0,0 +1,245 @@
#!/usr/bin/env python3
"""
Shared RFC 5424 syslog helper for DECNET service templates.
Provides two functions consumed by every service's server.py:
- syslog_line(service, hostname, event_type, severity, **fields) -> str
- write_syslog_file(line: str) -> None
- forward_syslog(line: str, log_target: str) -> None
RFC 5424 structure:
<PRI>1 TIMESTAMP HOSTNAME APP-NAME PROCID MSGID [SD-ELEMENT] MSG
Facility: local0 (16), PEN for SD element ID: decnet@55555
"""
import logging
import logging.handlers
import os
import socket
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
# ─── Constants ────────────────────────────────────────────────────────────────
_FACILITY_LOCAL0 = 16
_SD_ID = "decnet@55555"
_NILVALUE = "-"
SEVERITY_EMERG = 0
SEVERITY_ALERT = 1
SEVERITY_CRIT = 2
SEVERITY_ERROR = 3
SEVERITY_WARNING = 4
SEVERITY_NOTICE = 5
SEVERITY_INFO = 6
SEVERITY_DEBUG = 7
_MAX_HOSTNAME = 255
_MAX_APPNAME = 48
_MAX_MSGID = 32
_LOG_FILE_ENV = "DECNET_LOG_FILE"
_DEFAULT_LOG_FILE = "/var/log/decnet/decnet.log"
_MAX_BYTES = 10 * 1024 * 1024 # 10 MB
_BACKUP_COUNT = 5
# ─── Formatter ────────────────────────────────────────────────────────────────
def _sd_escape(value: str) -> str:
"""Escape SD-PARAM-VALUE per RFC 5424 §6.3.3."""
return value.replace("\\", "\\\\").replace('"', '\\"').replace("]", "\\]")
def _sd_element(fields: dict[str, Any]) -> str:
if not fields:
return _NILVALUE
params = " ".join(f'{k}="{_sd_escape(str(v))}"' for k, v in fields.items())
return f"[{_SD_ID} {params}]"
def syslog_line(
service: str,
hostname: str,
event_type: str,
severity: int = SEVERITY_INFO,
timestamp: datetime | None = None,
msg: str | None = None,
**fields: Any,
) -> str:
"""
Return a single RFC 5424-compliant syslog line (no trailing newline).
Args:
service: APP-NAME (e.g. "http", "mysql")
hostname: HOSTNAME (decky node name)
event_type: MSGID (e.g. "request", "login_attempt")
severity: Syslog severity integer (default: INFO=6)
timestamp: UTC datetime; defaults to now
msg: Optional free-text MSG
**fields: Encoded as structured data params
"""
pri = f"<{_FACILITY_LOCAL0 * 8 + severity}>"
ts = (timestamp or datetime.now(timezone.utc)).isoformat()
host = (hostname or _NILVALUE)[:_MAX_HOSTNAME]
appname = (service or _NILVALUE)[:_MAX_APPNAME]
msgid = (event_type or _NILVALUE)[:_MAX_MSGID]
sd = _sd_element(fields)
message = f" {msg}" if msg else ""
return f"{pri}1 {ts} {host} {appname} {_NILVALUE} {msgid} {sd}{message}"
# ─── File handler ─────────────────────────────────────────────────────────────
_file_logger: logging.Logger | None = None
def _get_file_logger() -> logging.Logger:
global _file_logger
if _file_logger is not None:
return _file_logger
log_path = Path(os.environ.get(_LOG_FILE_ENV, _DEFAULT_LOG_FILE))
try:
log_path.parent.mkdir(parents=True, exist_ok=True)
handler = logging.handlers.RotatingFileHandler(
log_path,
maxBytes=_MAX_BYTES,
backupCount=_BACKUP_COUNT,
encoding="utf-8",
)
except OSError:
handler = logging.StreamHandler()
handler.setFormatter(logging.Formatter("%(message)s"))
_file_logger = logging.getLogger("decnet.syslog")
_file_logger.setLevel(logging.DEBUG)
_file_logger.propagate = False
_file_logger.addHandler(handler)
return _file_logger
_json_logger: logging.Logger | None = None
def _get_json_logger() -> logging.Logger:
global _json_logger
if _json_logger is not None:
return _json_logger
log_path_str = os.environ.get(_LOG_FILE_ENV, _DEFAULT_LOG_FILE)
json_path = Path(log_path_str).with_suffix(".json")
try:
json_path.parent.mkdir(parents=True, exist_ok=True)
handler = logging.handlers.RotatingFileHandler(
json_path,
maxBytes=_MAX_BYTES,
backupCount=_BACKUP_COUNT,
encoding="utf-8",
)
except OSError:
handler = logging.StreamHandler()
handler.setFormatter(logging.Formatter("%(message)s"))
_json_logger = logging.getLogger("decnet.json")
_json_logger.setLevel(logging.DEBUG)
_json_logger.propagate = False
_json_logger.addHandler(handler)
return _json_logger
def write_syslog_file(line: str) -> None:
"""Append a syslog line to the rotating log file."""
try:
_get_file_logger().info(line)
# Also parse and write JSON log
import json
import re
from datetime import datetime
from typing import Optional, Any
_RFC5424_RE: re.Pattern = re.compile(
r"^<\d+>1 "
r"(\S+) " # 1: TIMESTAMP
r"(\S+) " # 2: HOSTNAME (decky name)
r"(\S+) " # 3: APP-NAME (service)
r"- " # PROCID always NILVALUE
r"(\S+) " # 4: MSGID (event_type)
r"(.+)$", # 5: SD element + optional MSG
)
_SD_BLOCK_RE: re.Pattern = re.compile(r'\[decnet@55555\s+(.*?)\]', re.DOTALL)
_PARAM_RE: re.Pattern = re.compile(r'(\w+)="((?:[^"\\]|\\.)*)"')
_IP_FIELDS: tuple[str, ...] = ("src_ip", "src", "client_ip", "remote_ip", "ip")
_m: Optional[re.Match] = _RFC5424_RE.match(line)
if _m:
_ts_raw: str
_decky: str
_service: str
_event_type: str
_sd_rest: str
_ts_raw, _decky, _service, _event_type, _sd_rest = _m.groups()
_fields: dict[str, str] = {}
_msg: str = ""
if _sd_rest.startswith("-"):
_msg = _sd_rest[1:].lstrip()
elif _sd_rest.startswith("["):
_block: Optional[re.Match] = _SD_BLOCK_RE.search(_sd_rest)
if _block:
for _k, _v in _PARAM_RE.findall(_block.group(1)):
_fields[_k] = _v.replace('\\"', '"').replace("\\\\", "\\").replace("\\]", "]")
# extract msg after the block
_msg_match: Optional[re.Match] = re.search(r'\]\s+(.+)$', _sd_rest)
if _msg_match:
_msg = _msg_match.group(1).strip()
else:
_msg = _sd_rest
_attacker_ip: str = "Unknown"
for _fname in _IP_FIELDS:
if _fname in _fields:
_attacker_ip = _fields[_fname]
break
# Parse timestamp to normalize it
_ts_formatted: str
try:
_ts_formatted = datetime.fromisoformat(_ts_raw).strftime("%Y-%m-%d %H:%M:%S")
except ValueError:
_ts_formatted = _ts_raw
_payload: dict[str, Any] = {
"timestamp": _ts_formatted,
"decky": _decky,
"service": _service,
"event_type": _event_type,
"attacker_ip": _attacker_ip,
"fields": json.dumps(_fields),
"msg": _msg,
"raw_line": line
}
_get_json_logger().info(json.dumps(_payload))
except Exception:
pass
# ─── TCP forwarding ───────────────────────────────────────────────────────────
def forward_syslog(line: str, log_target: str) -> None:
"""Forward a syslog line over TCP to log_target (ip:port)."""
if not log_target:
return
try:
host, port = log_target.rsplit(":", 1)
with socket.create_connection((host, int(port)), timeout=3) as s:
s.sendall((line + "\n").encode())
except Exception:
pass

2
templates/smtp/server.py
View File

@@ -95,7 +95,7 @@ class SMTPProtocol(asyncio.Protocol):
async def main():
_log("startup", msg=f"SMTP server starting as {NODE_NAME}")
loop = asyncio.get_running_loop()
server = await loop.create_server(SMTPProtocol, "0.0.0.0", 25)
server = await loop.create_server(SMTPProtocol, "0.0.0.0", 25) # nosec B104
async with server:
await server.serve_forever()

16
templates/snmp/server.py
View File

@@ -91,17 +91,17 @@ def _ber_tlv(tag: int, value: bytes) -> bytes:
def _parse_snmp(data: bytes):
"""Return (version, community, request_id, oids) or raise."""
pos = 0
assert data[pos] == 0x30
assert data[pos] == 0x30 # nosec B101
pos += 1
_, pos = _read_ber_length(data, pos)
# version
assert data[pos] == 0x02
assert data[pos] == 0x02 # nosec B101
pos += 1
v_len, pos = _read_ber_length(data, pos)
version = int.from_bytes(data[pos:pos + v_len], "big")
pos += v_len
# community
assert data[pos] == 0x04
assert data[pos] == 0x04 # nosec B101
pos += 1
c_len, pos = _read_ber_length(data, pos)
community = data[pos:pos + c_len].decode(errors="replace")
@@ -110,23 +110,23 @@ def _parse_snmp(data: bytes):
pos += 1
_, pos = _read_ber_length(data, pos)
# request-id
assert data[pos] == 0x02
assert data[pos] == 0x02 # nosec B101
pos += 1
r_len, pos = _read_ber_length(data, pos)
request_id = int.from_bytes(data[pos:pos + r_len], "big")
pos += r_len
pos += 4 # skip error-status and error-index
# varbind list
assert data[pos] == 0x30
assert data[pos] == 0x30 # nosec B101
pos += 1
vbl_len, pos = _read_ber_length(data, pos)
end = pos + vbl_len
oids = []
while pos < end:
assert data[pos] == 0x30
assert data[pos] == 0x30 # nosec B101
pos += 1
vb_len, pos = _read_ber_length(data, pos)
assert data[pos] == 0x06
assert data[pos] == 0x06 # nosec B101
pos += 1
oid_len, pos = _read_ber_length(data, pos)
oid = _decode_oid(data[pos:pos + oid_len])
@@ -179,7 +179,7 @@ async def main():
_log("startup", msg=f"SNMP server starting as {NODE_NAME}")
loop = asyncio.get_running_loop()
transport, _ = await loop.create_datagram_endpoint(
SNMPProtocol, local_addr=("0.0.0.0", 161)
SNMPProtocol, local_addr=("0.0.0.0", 161) # nosec B104
)
try:
await asyncio.sleep(float("inf"))

2
templates/tftp/server.py
View File

@@ -69,7 +69,7 @@ async def main():
_log("startup", msg=f"TFTP server starting as {NODE_NAME}")
loop = asyncio.get_running_loop()
transport, _ = await loop.create_datagram_endpoint(
TFTPProtocol, local_addr=("0.0.0.0", 69)
TFTPProtocol, local_addr=("0.0.0.0", 69) # nosec B104
)
try:
await asyncio.sleep(float("inf"))

2
templates/vnc/server.py
View File

@@ -87,7 +87,7 @@ class VNCProtocol(asyncio.Protocol):
async def main():
_log("startup", msg=f"VNC server starting as {NODE_NAME}")
loop = asyncio.get_running_loop()
server = await loop.create_server(VNCProtocol, "0.0.0.0", 5900)
server = await loop.create_server(VNCProtocol, "0.0.0.0", 5900) # nosec B104
async with server:
await server.serve_forever()