diff --git a/ansible/deploy_aw_server.yml b/ansible/deploy_aw_server.yml index fc93e03..96826b8 100644 --- a/ansible/deploy_aw_server.yml +++ b/ansible/deploy_aw_server.yml @@ -409,8 +409,10 @@ mode: "0644" loop: - cef_exporter.py + - syslog_forwarder.py - webhook_sender.py - cef-config.yaml + - syslog-forwarder-config.yaml - webhook-config.yaml - requirements.txt when: aw_dlp_integrations_enabled | default(true) | bool @@ -423,6 +425,7 @@ mode: "0755" loop: - cef_exporter.py + - syslog_forwarder.py - webhook_sender.py when: aw_dlp_integrations_enabled | default(true) | bool @@ -466,6 +469,24 @@ mode: "0644" when: aw_dlp_integrations_enabled | default(true) | bool + - name: Установить systemd unit syslog forwarder + ansible.builtin.copy: + src: "{{ aw_repo_root }}/aw-server/dlp-integrations/syslog-forwarder.service" + dest: /etc/systemd/system/aw-dlp-syslog-forwarder.service + owner: root + group: root + mode: "0644" + when: aw_dlp_integrations_enabled | default(true) | bool + + - name: Установить systemd timer syslog forwarder + ansible.builtin.copy: + src: "{{ aw_repo_root }}/aw-server/dlp-integrations/syslog-forwarder.timer" + dest: /etc/systemd/system/aw-dlp-syslog-forwarder.timer + owner: root + group: root + mode: "0644" + when: aw_dlp_integrations_enabled | default(true) | bool + - name: Установить systemd unit webhook sender ansible.builtin.copy: src: "{{ aw_repo_root }}/aw-server/dlp-integrations/webhook-sender.service" @@ -504,6 +525,7 @@ - case_service.py - case_schema.py - case_storage.py + - evidence_chain.py - requirements.txt when: aw_dlp_case_management_enabled | default(true) | bool @@ -559,8 +581,10 @@ mode: "{{ item.mode }}" loop: - { src: "report_generator.py", dest: "/opt/activitywatch/dlp-compliance/report_generator.py", mode: "0755" } + - { src: "compliance_scheduler.py", dest: "/opt/activitywatch/dlp-compliance/compliance_scheduler.py", mode: "0755" } - { src: "requirements.txt", dest: "/opt/activitywatch/dlp-compliance/requirements.txt", mode: "0644" } - { src: "templates/152-fz-report.html", dest: "/opt/activitywatch/dlp-compliance/templates/152-fz-report.html", mode: "0644" } + - { src: "templates/pci-dss-report.html", dest: "/opt/activitywatch/dlp-compliance/templates/pci-dss-report.html", mode: "0644" } - { src: "report-scheduler.service", dest: "/etc/systemd/system/aw-dlp-report-scheduler.service", mode: "0644" } - { src: "report-scheduler.timer", dest: "/etc/systemd/system/aw-dlp-report-scheduler.timer", mode: "0644" } when: aw_dlp_compliance_enabled | default(true) | bool @@ -676,6 +700,13 @@ state: restarted when: aw_dlp_integrations_enabled | default(true) | bool + - name: Включить и перезапустить timer syslog forwarder + ansible.builtin.systemd: + name: aw-dlp-syslog-forwarder.timer + enabled: true + state: restarted + when: aw_dlp_integrations_enabled | default(true) | bool + - name: Включить и перезапустить timer webhook sender ansible.builtin.systemd: name: aw-dlp-webhook-sender.timer diff --git a/ansible/roles/dlp-case-management/tasks/main.yml b/ansible/roles/dlp-case-management/tasks/main.yml index ba06453..0b121a5 100644 --- a/ansible/roles/dlp-case-management/tasks/main.yml +++ b/ansible/roles/dlp-case-management/tasks/main.yml @@ -18,6 +18,7 @@ - case_service.py - case_schema.py - case_storage.py + - evidence_chain.py - requirements.txt - name: Make case service executable diff --git a/ansible/roles/dlp-compliance/tasks/main.yml b/ansible/roles/dlp-compliance/tasks/main.yml index 03faedf..3f681de 100644 --- a/ansible/roles/dlp-compliance/tasks/main.yml +++ b/ansible/roles/dlp-compliance/tasks/main.yml @@ -20,8 +20,10 @@ mode: "{{ item.mode }}" loop: - { src: "report_generator.py", dest: "/opt/activitywatch/dlp-compliance/report_generator.py", mode: "0755" } + - { src: "compliance_scheduler.py", dest: "/opt/activitywatch/dlp-compliance/compliance_scheduler.py", mode: "0755" } - { src: "requirements.txt", dest: "/opt/activitywatch/dlp-compliance/requirements.txt", mode: "0644" } - { src: "templates/152-fz-report.html", dest: "/opt/activitywatch/dlp-compliance/templates/152-fz-report.html", mode: "0644" } + - { src: "templates/pci-dss-report.html", dest: "/opt/activitywatch/dlp-compliance/templates/pci-dss-report.html", mode: "0644" } - { src: "report-scheduler.service", dest: "/etc/systemd/system/aw-dlp-report-scheduler.service", mode: "0644" } - { src: "report-scheduler.timer", dest: "/etc/systemd/system/aw-dlp-report-scheduler.timer", mode: "0644" } @@ -50,4 +52,3 @@ ansible.builtin.systemd: name: aw-dlp-report-scheduler.service state: started - diff --git a/ansible/roles/dlp-content-analysis/tasks/main.yml b/ansible/roles/dlp-content-analysis/tasks/main.yml new file mode 100644 index 0000000..6a8dde1 --- /dev/null +++ b/ansible/roles/dlp-content-analysis/tasks/main.yml @@ -0,0 +1,37 @@ +--- +- name: Install OCR/system dependencies + ansible.builtin.apt: + name: + - python3-venv + - python3-pip + - tesseract-ocr + - tesseract-ocr-rus + state: present + update_cache: true + +- name: Create content analysis directory + ansible.builtin.file: + path: /opt/activitywatch/dlp-content-analysis + state: directory + owner: "{{ aw_server_user | default('activitywatch') }}" + group: "{{ aw_server_group | default('activitywatch') }}" + mode: "0755" + +- name: Copy content analysis sources + ansible.builtin.copy: + src: "{{ playbook_dir }}/../aw-server/dlp-content-analysis/" + dest: /opt/activitywatch/dlp-content-analysis/ + owner: "{{ aw_server_user | default('activitywatch') }}" + group: "{{ aw_server_group | default('activitywatch') }}" + mode: "0644" + +- name: Create venv for content analysis + ansible.builtin.command: + cmd: python3 -m venv /opt/activitywatch/dlp-content-analysis/.venv + args: + creates: /opt/activitywatch/dlp-content-analysis/.venv/bin/python + +- name: Install Python dependencies for content analysis + ansible.builtin.pip: + requirements: /opt/activitywatch/dlp-content-analysis/requirements.txt + virtualenv: /opt/activitywatch/dlp-content-analysis/.venv diff --git a/ansible/roles/dlp-integrations/tasks/main.yml b/ansible/roles/dlp-integrations/tasks/main.yml index 9676b13..ec1e3bd 100644 --- a/ansible/roles/dlp-integrations/tasks/main.yml +++ b/ansible/roles/dlp-integrations/tasks/main.yml @@ -16,8 +16,10 @@ mode: "0644" loop: - cef_exporter.py + - syslog_forwarder.py - webhook_sender.py - cef-config.yaml + - syslog-forwarder-config.yaml - webhook-config.yaml - requirements.txt @@ -29,6 +31,7 @@ mode: "0755" loop: - cef_exporter.py + - syslog_forwarder.py - webhook_sender.py - name: Create venv for DLP integrations @@ -60,6 +63,8 @@ loop: - { src: "cef-exporter.service", dest: "/etc/systemd/system/aw-dlp-cef-exporter.service" } - { src: "cef-exporter.timer", dest: "/etc/systemd/system/aw-dlp-cef-exporter.timer" } + - { src: "syslog-forwarder.service", dest: "/etc/systemd/system/aw-dlp-syslog-forwarder.service" } + - { src: "syslog-forwarder.timer", dest: "/etc/systemd/system/aw-dlp-syslog-forwarder.timer" } - { src: "webhook-sender.service", dest: "/etc/systemd/system/aw-dlp-webhook-sender.service" } - { src: "webhook-sender.timer", dest: "/etc/systemd/system/aw-dlp-webhook-sender.timer" } @@ -74,4 +79,5 @@ state: restarted loop: - aw-dlp-cef-exporter.timer + - aw-dlp-syslog-forwarder.timer - aw-dlp-webhook-sender.timer diff --git a/aw-server/dlp-case-management/case_storage.py b/aw-server/dlp-case-management/case_storage.py index 5fab597..429be99 100644 --- a/aw-server/dlp-case-management/case_storage.py +++ b/aw-server/dlp-case-management/case_storage.py @@ -8,6 +8,8 @@ from datetime import datetime, timezone from pathlib import Path from typing import Any, Iterator +from evidence_chain import evidence_sha256, normalize_evidence_chain + class CaseStorage: def __init__(self, db_path: Path) -> None: @@ -98,6 +100,15 @@ class CaseStorage: def create_case(self, payload: dict[str, Any], actor: str | None = None) -> dict[str, Any]: now = self._now() + normalized_evidence = None + evidence_digest = None + if payload.get("evidence") is not None: + normalized_evidence = normalize_evidence_chain( + payload=payload.get("evidence"), + source_bucket=payload.get("source_bucket"), + source_event_ts=payload.get("source_event_ts"), + ) + evidence_digest = normalized_evidence.get("latest_sha256") or evidence_sha256(payload.get("evidence")) with self.conn() as c: cur = c.execute( """ @@ -114,7 +125,7 @@ class CaseStorage: payload.get("assignee"), payload.get("source_bucket"), payload.get("source_event_ts"), - json.dumps(payload.get("evidence"), ensure_ascii=False) if payload.get("evidence") is not None else None, + json.dumps(normalized_evidence, ensure_ascii=False) if normalized_evidence is not None else None, now, now, ), @@ -125,7 +136,10 @@ class CaseStorage: case_id=case_id, action="create", actor=actor, - details={"fields": {k: v for k, v in payload.items() if k != "evidence"}}, + details={ + "fields": {k: v for k, v in payload.items() if k != "evidence"}, + "evidence_sha256": evidence_digest, + }, ) c.commit() return self.get_case(case_id, c) @@ -252,4 +266,3 @@ class CaseStorage: self._now(), ), ) - diff --git a/aw-server/dlp-case-management/evidence_chain.py b/aw-server/dlp-case-management/evidence_chain.py new file mode 100644 index 0000000..c8c4294 --- /dev/null +++ b/aw-server/dlp-case-management/evidence_chain.py @@ -0,0 +1,52 @@ +#!/usr/bin/env python3 +from __future__ import annotations + +import hashlib +import json +from datetime import datetime, timezone +from typing import Any + + +def _utc_now() -> str: + return datetime.now(timezone.utc).isoformat() + + +def _canonical_json(payload: Any) -> str: + return json.dumps(payload, ensure_ascii=False, sort_keys=True, separators=(",", ":")) + + +def evidence_sha256(payload: Any) -> str: + return hashlib.sha256(_canonical_json(payload).encode("utf-8")).hexdigest() + + +def build_evidence_record( + payload: Any, + source_bucket: str | None = None, + source_event_ts: str | None = None, +) -> dict[str, Any]: + return { + "recorded_at": _utc_now(), + "source_bucket": source_bucket, + "source_event_ts": source_event_ts, + "sha256": evidence_sha256(payload), + "payload": payload, + } + + +def normalize_evidence_chain( + payload: Any, + source_bucket: str | None = None, + source_event_ts: str | None = None, +) -> dict[str, Any]: + if isinstance(payload, dict) and isinstance(payload.get("items"), list): + return payload + record = build_evidence_record( + payload=payload, + source_bucket=source_bucket, + source_event_ts=source_event_ts, + ) + return { + "items": [record], + "latest_sha256": record["sha256"], + "chain_length": 1, + } diff --git a/aw-server/dlp-compliance/compliance_scheduler.py b/aw-server/dlp-compliance/compliance_scheduler.py new file mode 100644 index 0000000..fb55a9a --- /dev/null +++ b/aw-server/dlp-compliance/compliance_scheduler.py @@ -0,0 +1,24 @@ +#!/usr/bin/env python3 +from __future__ import annotations + +import argparse +import json + +from report_generator import generate_report + + +def main() -> None: + parser = argparse.ArgumentParser(description="Run one or more DLP compliance report profiles") + parser.add_argument("--month", help="Month in YYYY-MM format") + parser.add_argument("--profiles", default="152-fz,pci-dss", help="Comma-separated profiles to generate") + parser.add_argument("--stdout-json", action="store_true") + args = parser.parse_args() + + profiles = [item.strip() for item in str(args.profiles).split(",") if item.strip()] + results = [generate_report(month=args.month, profile=profile) for profile in profiles] + if args.stdout_json: + print(json.dumps({"items": results}, ensure_ascii=False)) + + +if __name__ == "__main__": + main() diff --git a/aw-server/dlp-compliance/report-scheduler.service b/aw-server/dlp-compliance/report-scheduler.service index a9395a3..d0ad7da 100644 --- a/aw-server/dlp-compliance/report-scheduler.service +++ b/aw-server/dlp-compliance/report-scheduler.service @@ -6,7 +6,6 @@ After=network-online.target Type=oneshot EnvironmentFile=-/etc/activitywatch/aw-server.env WorkingDirectory=/opt/activitywatch/dlp-compliance -ExecStart=/opt/activitywatch/dlp-compliance/.venv/bin/python /opt/activitywatch/dlp-compliance/report_generator.py +ExecStart=/opt/activitywatch/dlp-compliance/.venv/bin/python /opt/activitywatch/dlp-compliance/compliance_scheduler.py User=activitywatch Group=activitywatch - diff --git a/aw-server/dlp-compliance/report_generator.py b/aw-server/dlp-compliance/report_generator.py index 0a502f1..9e8fad7 100644 --- a/aw-server/dlp-compliance/report_generator.py +++ b/aw-server/dlp-compliance/report_generator.py @@ -18,7 +18,11 @@ def _env(name: str, default: str) -> str: AW_API_BASE = _env("AW_SERVER_URL", "http://127.0.0.1:5600/api/0").rstrip("/") OUTPUT_DIR = Path(_env("AW_DLP_COMPLIANCE_REPORT_DIR", "/opt/activitywatch/dlp-compliance/reports")) -TEMPLATE_PATH = Path(_env("AW_DLP_COMPLIANCE_TEMPLATE", "/opt/activitywatch/dlp-compliance/templates/152-fz-report.html")) +BASE_DIR = Path(__file__).resolve().parent +PROFILE_TEMPLATE_MAP = { + "152-fz": BASE_DIR / "templates" / "152-fz-report.html", + "pci-dss": BASE_DIR / "templates" / "pci-dss-report.html", +} @dataclass @@ -108,10 +112,18 @@ def _render_table(title: str, rows: list[tuple[str, int]]) -> str: return f"

{title}

{body}
ПараметрЗначение
" -def _render_html(period_label: str, stats: ReportStats, generated_at: str) -> str: - template = TEMPLATE_PATH.read_text(encoding="utf-8") +def _resolve_template_path(profile: str) -> Path: + if profile == "152-fz": + explicit = _env("AW_DLP_COMPLIANCE_TEMPLATE", str(PROFILE_TEMPLATE_MAP["152-fz"])) + return Path(explicit) + return PROFILE_TEMPLATE_MAP.get(profile, PROFILE_TEMPLATE_MAP["152-fz"]) + + +def _render_html(profile: str, period_label: str, stats: ReportStats, generated_at: str) -> str: + template = _resolve_template_path(profile).read_text(encoding="utf-8") return ( template.replace("{{PERIOD}}", period_label) + .replace("{{PROFILE}}", profile) .replace("{{GENERATED_AT}}", generated_at) .replace("{{TOTAL}}", str(stats.total_incidents)) .replace("{{HIGH}}", str(stats.high)) @@ -136,22 +148,18 @@ def _period_bounds(month: str | None) -> tuple[datetime, datetime, str]: return start, end, start.strftime("%Y-%m") -def main() -> None: - parser = argparse.ArgumentParser(description="Generate 152-FZ compliance report from AW DLP incidents") - parser.add_argument("--month", help="Month in YYYY-MM format (default: current month)") - parser.add_argument("--stdout-json", action="store_true", help="Print report metadata as JSON") - args = parser.parse_args() - - start, end, period_label = _period_bounds(args.month) +def generate_report(month: str | None = None, profile: str = "152-fz") -> dict[str, object]: + start, end, period_label = _period_bounds(month) incidents = _load_incidents(start, end) stats = _build_stats(incidents) OUTPUT_DIR.mkdir(parents=True, exist_ok=True) generated_at = datetime.now(UTC).isoformat().replace("+00:00", "Z") - html_out = OUTPUT_DIR / f"152-fz-{period_label}.html" - html_out.write_text(_render_html(period_label, stats, generated_at), encoding="utf-8") + html_out = OUTPUT_DIR / f"{profile}-{period_label}.html" + html_out.write_text(_render_html(profile, period_label, stats, generated_at), encoding="utf-8") metadata = { + "profile": profile, "period": period_label, "generated_at": generated_at, "aw_api_base": AW_API_BASE, @@ -163,10 +171,20 @@ def main() -> None: "low": stats.low, }, } - (OUTPUT_DIR / f"152-fz-{period_label}.json").write_text( + (OUTPUT_DIR / f"{profile}-{period_label}.json").write_text( json.dumps(metadata, ensure_ascii=False, indent=2), encoding="utf-8", ) + return metadata + + +def main() -> None: + parser = argparse.ArgumentParser(description="Generate DLP compliance report from AW DLP incidents") + parser.add_argument("--month", help="Month in YYYY-MM format (default: current month)") + parser.add_argument("--profile", default="152-fz", help="Profile name: 152-fz or pci-dss") + parser.add_argument("--stdout-json", action="store_true", help="Print report metadata as JSON") + args = parser.parse_args() + metadata = generate_report(month=args.month, profile=args.profile) if args.stdout_json: print(json.dumps(metadata, ensure_ascii=False)) diff --git a/aw-server/dlp-compliance/templates/pci-dss-report.html b/aw-server/dlp-compliance/templates/pci-dss-report.html new file mode 100644 index 0000000..f2ffce3 --- /dev/null +++ b/aw-server/dlp-compliance/templates/pci-dss-report.html @@ -0,0 +1,33 @@ + + + + + PCI DSS Report {{PERIOD}} + + + +

Compliance отчёт PCI DSS

+
Профиль: {{PROFILE}} | Период: {{PERIOD}} | Сформирован: {{GENERATED_AT}}
+ +
+
Всего инцидентов
{{TOTAL}}
+
High
{{HIGH}}
+
Medium
{{MEDIUM}}
+
Low
{{LOW}}
+
+ + {{HOST_TABLE}} + {{CHANNEL_TABLE}} + + diff --git a/aw-server/dlp-content-analysis/content_analyzer.py b/aw-server/dlp-content-analysis/content_analyzer.py new file mode 100644 index 0000000..c420916 --- /dev/null +++ b/aw-server/dlp-content-analysis/content_analyzer.py @@ -0,0 +1,98 @@ +#!/usr/bin/env python3 +from __future__ import annotations + +import argparse +import json +from pathlib import Path +from typing import Any + +from dictionary_matcher import match_text +from ocr_processor import analyze_screenshot + +BASE_DIR = Path(__file__).resolve().parent +DICTIONARY_DIR = BASE_DIR / "dictionaries" +REGEX_PACK_DIR = BASE_DIR / "regex-packs" + + +def resolve_dictionary_pack(name: str | None) -> str | None: + if not name: + return None + path = DICTIONARY_DIR / f"{name}.json" + return str(path) if path.exists() else None + + +def resolve_regex_pack(name: str | None) -> str | None: + if not name: + return None + path = REGEX_PACK_DIR / f"{name}.json" + return str(path) if path.exists() else None + + +def analyze_text_content( + text: str, + dictionary_pack: str | None = None, + regex_pack: str | None = None, +) -> dict[str, Any]: + dictionary_path = resolve_dictionary_pack(dictionary_pack) + regex_pack_path = resolve_regex_pack(regex_pack) + matches = match_text( + text=text, + dictionary_path=dictionary_path, + regex_pack_path=regex_pack_path, + ) + return { + "text": text, + "dictionary_pack": dictionary_pack, + "regex_pack": regex_pack, + "dictionary_matches": matches.get("dictionary_matches", []), + "regex_matches": matches.get("regex_matches", []), + } + + +def analyze_artifact( + text: str | None = None, + image_path: str | None = None, + dictionary_pack: str | None = None, + regex_pack: str | None = None, +) -> dict[str, Any]: + if image_path: + dictionary_path = resolve_dictionary_pack(dictionary_pack) + regex_pack_path = resolve_regex_pack(regex_pack) + result = analyze_screenshot( + image_path=image_path, + dictionary_path=dictionary_path, + regex_pack_path=regex_pack_path, + ) + result["dictionary_pack"] = dictionary_pack + result["regex_pack"] = regex_pack + result["source"] = "image" + return result + return { + **analyze_text_content( + text=text or "", + dictionary_pack=dictionary_pack, + regex_pack=regex_pack, + ), + "source": "text", + } + + +def main() -> None: + parser = argparse.ArgumentParser(description="Analyze text or screenshot with DLP dictionaries/regex packs") + parser.add_argument("--text", help="Text to analyze") + parser.add_argument("--image", help="Screenshot/image path to analyze") + parser.add_argument("--dictionary-pack", default=None) + parser.add_argument("--regex-pack", default=None) + args = parser.parse_args() + + result = analyze_artifact( + text=args.text, + image_path=args.image, + dictionary_pack=args.dictionary_pack, + regex_pack=args.regex_pack, + ) + print(json.dumps(result, ensure_ascii=False, indent=2)) + + +if __name__ == "__main__": + main() diff --git a/aw-server/dlp-content-analysis/requirements.txt b/aw-server/dlp-content-analysis/requirements.txt new file mode 100644 index 0000000..4939f3f --- /dev/null +++ b/aw-server/dlp-content-analysis/requirements.txt @@ -0,0 +1,2 @@ +pytesseract>=0.3.10 +Pillow>=10.0.0 diff --git a/aw-server/dlp-integrations/syslog-forwarder-config.yaml b/aw-server/dlp-integrations/syslog-forwarder-config.yaml new file mode 100644 index 0000000..b744f90 --- /dev/null +++ b/aw-server/dlp-integrations/syslog-forwarder-config.yaml @@ -0,0 +1,8 @@ +aw_api_base: "http://127.0.0.1:5600/api/0" +state_path: "/var/lib/activitywatch/dlp-integrations/syslog-forwarder-state.json" +syslog_host: "127.0.0.1" +syslog_port: 514 +syslog_proto: "udp" +facility: 16 +app_name: "aw-dlp" +per_bucket_limit: 300 diff --git a/aw-server/dlp-integrations/syslog-forwarder.service b/aw-server/dlp-integrations/syslog-forwarder.service new file mode 100644 index 0000000..d354551 --- /dev/null +++ b/aw-server/dlp-integrations/syslog-forwarder.service @@ -0,0 +1,10 @@ +[Unit] +Description=AWatch DLP syslog forwarder +After=network-online.target + +[Service] +Type=oneshot +WorkingDirectory=/opt/activitywatch/dlp-integrations +ExecStart=/opt/activitywatch/dlp-integrations/.venv/bin/python /opt/activitywatch/dlp-integrations/syslog_forwarder.py +User=activitywatch +Group=activitywatch diff --git a/aw-server/dlp-integrations/syslog-forwarder.timer b/aw-server/dlp-integrations/syslog-forwarder.timer new file mode 100644 index 0000000..218a308 --- /dev/null +++ b/aw-server/dlp-integrations/syslog-forwarder.timer @@ -0,0 +1,11 @@ +[Unit] +Description=Run AWatch DLP syslog forwarder every 3 minutes + +[Timer] +OnBootSec=3min +OnUnitActiveSec=3min +Persistent=true +Unit=aw-dlp-syslog-forwarder.service + +[Install] +WantedBy=timers.target diff --git a/aw-server/dlp-integrations/syslog_forwarder.py b/aw-server/dlp-integrations/syslog_forwarder.py new file mode 100644 index 0000000..8c32616 --- /dev/null +++ b/aw-server/dlp-integrations/syslog_forwarder.py @@ -0,0 +1,142 @@ +#!/usr/bin/env python3 +from __future__ import annotations + +import json +import logging +import socket +from datetime import datetime, timezone +from pathlib import Path +from typing import Any +from urllib import error, request + +import yaml + +LOG = logging.getLogger("aw.dlp.syslog_forwarder") + + +def setup_logging() -> None: + logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") + + +def load_yaml(path: Path) -> dict[str, Any]: + if not path.exists(): + return {} + data = yaml.safe_load(path.read_text(encoding="utf-8")) or {} + return data if isinstance(data, dict) else {} + + +def load_json(path: Path) -> dict[str, Any]: + if not path.exists(): + return {} + try: + data = json.loads(path.read_text(encoding="utf-8")) + except Exception: + return {} + return data if isinstance(data, dict) else {} + + +def save_json(path: Path, payload: dict[str, Any]) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8") + + +def http_json(url: str, timeout: int = 15) -> Any: + req = request.Request(url, method="GET") + with request.urlopen(req, timeout=timeout) as resp: + return json.loads(resp.read().decode("utf-8", errors="ignore")) + + +def iter_new_incidents(aw_base: str, state: dict[str, Any], per_bucket_limit: int) -> tuple[list[dict[str, Any]], dict[str, int]]: + buckets = http_json(f"{aw_base}/buckets/") + bucket_ids = sorted([bid for bid in buckets.keys() if bid.startswith("aw-dlp-incidents_")]) + last_ids = state.get("last_ids", {}) + if not isinstance(last_ids, dict): + last_ids = {} + max_ids: dict[str, int] = {} + out: list[dict[str, Any]] = [] + for bid in bucket_ids: + try: + events = http_json(f"{aw_base}/buckets/{bid}/events?limit={int(per_bucket_limit)}") + except error.HTTPError as exc: + LOG.warning("skip bucket %s: %s", bid, exc) + continue + prev = int(last_ids.get(bid, 0)) + bucket_max = prev + for ev in events: + eid = int(ev.get("id") or 0) + if eid <= prev: + continue + out.append(ev) + if eid > bucket_max: + bucket_max = eid + max_ids[bid] = bucket_max + out.sort(key=lambda x: int(x.get("id") or 0)) + return out, max_ids + + +def build_message(event: dict[str, Any], app_name: str, facility: int) -> str: + pri = facility * 8 + 6 + ts = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z") + data = event.get("data") or {} + host = str(data.get("hostname") or "unknown") + payload = json.dumps( + { + "event_id": event.get("id"), + "timestamp": event.get("timestamp"), + "host": host, + "severity": data.get("severity"), + "signalType": data.get("signalType"), + "username": data.get("username"), + "action": data.get("action"), + "message": data.get("message"), + "data": data, + }, + ensure_ascii=False, + separators=(",", ":"), + ) + return f"<{pri}>1 {ts} {host} {app_name} - - - {payload}" + + +def send_syslog(line: str, host: str, port: int, proto: str, timeout: int = 10) -> None: + if proto.lower() == "tcp": + sock = socket.create_connection((host, port), timeout=timeout) + try: + sock.sendall((line + "\n").encode("utf-8", errors="ignore")) + finally: + sock.close() + return + sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + try: + sock.sendto(line.encode("utf-8", errors="ignore"), (host, port)) + finally: + sock.close() + + +def main() -> None: + setup_logging() + cfg_path = Path("/opt/activitywatch/dlp-integrations/syslog-forwarder-config.yaml") + cfg = load_yaml(cfg_path) + aw_base = str(cfg.get("aw_api_base", "http://127.0.0.1:5600/api/0")).rstrip("/") + state_path = Path(str(cfg.get("state_path", "/var/lib/activitywatch/dlp-integrations/syslog-forwarder-state.json"))) + per_bucket_limit = int(cfg.get("per_bucket_limit", 300)) + syslog_host = str(cfg.get("syslog_host", "127.0.0.1")) + syslog_port = int(cfg.get("syslog_port", 514)) + syslog_proto = str(cfg.get("syslog_proto", "udp")) + facility = int(cfg.get("facility", 16)) + app_name = str(cfg.get("app_name", "aw-dlp")) + + state = load_json(state_path) + incidents, max_ids = iter_new_incidents(aw_base=aw_base, state=state, per_bucket_limit=per_bucket_limit) + + sent = 0 + for event in incidents: + line = build_message(event, app_name=app_name, facility=facility) + send_syslog(line=line, host=syslog_host, port=syslog_port, proto=syslog_proto) + sent += 1 + + save_json(state_path, {"last_ids": max_ids}) + LOG.info("syslog forwarder sent=%d buckets=%d", sent, len(max_ids)) + + +if __name__ == "__main__": + main()