diff --git a/ansible/deploy_aw_server.yml b/ansible/deploy_aw_server.yml index 79677ae..6f47ba1 100644 --- a/ansible/deploy_aw_server.yml +++ b/ansible/deploy_aw_server.yml @@ -385,6 +385,99 @@ virtualenv: /opt/activitywatch/dlp-content-analysis/.venv when: aw_dlp_content_analysis_enabled | default(true) | bool + - name: Создать каталог DLP integrations + ansible.builtin.file: + path: /opt/activitywatch/dlp-integrations + state: directory + owner: "{{ aw_server_user }}" + group: "{{ aw_server_group }}" + mode: "0755" + when: aw_dlp_integrations_enabled | default(true) | bool + + - name: Скопировать файлы DLP integrations + ansible.builtin.copy: + src: "{{ aw_repo_root }}/aw-server/dlp-integrations/{{ item }}" + dest: "/opt/activitywatch/dlp-integrations/{{ item }}" + owner: "{{ aw_server_user }}" + group: "{{ aw_server_group }}" + mode: "0644" + loop: + - cef_exporter.py + - webhook_sender.py + - cef-config.yaml + - webhook-config.yaml + - requirements.txt + when: aw_dlp_integrations_enabled | default(true) | bool + + - name: Выставить executable для python scripts DLP integrations + ansible.builtin.file: + path: "/opt/activitywatch/dlp-integrations/{{ item }}" + owner: "{{ aw_server_user }}" + group: "{{ aw_server_group }}" + mode: "0755" + loop: + - cef_exporter.py + - webhook_sender.py + when: aw_dlp_integrations_enabled | default(true) | bool + + - name: Создать virtualenv DLP integrations + ansible.builtin.command: + cmd: python3 -m venv /opt/activitywatch/dlp-integrations/.venv + args: + creates: /opt/activitywatch/dlp-integrations/.venv/bin/python + when: aw_dlp_integrations_enabled | default(true) | bool + + - name: Установить зависимости DLP integrations + ansible.builtin.pip: + requirements: /opt/activitywatch/dlp-integrations/requirements.txt + virtualenv: /opt/activitywatch/dlp-integrations/.venv + when: aw_dlp_integrations_enabled | default(true) | bool + + - name: Создать state каталог DLP integrations + ansible.builtin.file: + path: /var/lib/activitywatch/dlp-integrations + state: directory + owner: "{{ aw_server_user }}" + group: "{{ aw_server_group }}" + mode: "0755" + when: aw_dlp_integrations_enabled | default(true) | bool + + - name: Установить systemd unit CEF exporter + ansible.builtin.copy: + src: "{{ aw_repo_root }}/aw-server/dlp-integrations/cef-exporter.service" + dest: /etc/systemd/system/aw-dlp-cef-exporter.service + owner: root + group: root + mode: "0644" + when: aw_dlp_integrations_enabled | default(true) | bool + + - name: Установить systemd timer CEF exporter + ansible.builtin.copy: + src: "{{ aw_repo_root }}/aw-server/dlp-integrations/cef-exporter.timer" + dest: /etc/systemd/system/aw-dlp-cef-exporter.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" + dest: /etc/systemd/system/aw-dlp-webhook-sender.service + owner: root + group: root + mode: "0644" + when: aw_dlp_integrations_enabled | default(true) | bool + + - name: Установить systemd timer webhook sender + ansible.builtin.copy: + src: "{{ aw_repo_root }}/aw-server/dlp-integrations/webhook-sender.timer" + dest: /etc/systemd/system/aw-dlp-webhook-sender.timer + owner: root + group: root + mode: "0644" + when: aw_dlp_integrations_enabled | default(true) | bool + - name: Установить скрипт AW worktime API ansible.builtin.copy: src: "{{ aw_repo_root }}/aw-server/aw-worktime-api.py" @@ -468,6 +561,20 @@ state: restarted when: aw_dlp_policy_engine_enabled | default(false) | bool + - name: Включить и перезапустить timer CEF exporter + ansible.builtin.systemd: + name: aw-dlp-cef-exporter.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 + enabled: true + state: restarted + when: aw_dlp_integrations_enabled | default(true) | bool + - name: Включить и перезапустить AW worktime API ansible.builtin.systemd: name: aw-worktime-api.service diff --git a/ansible/deploy_dlp_full_stack.yml b/ansible/deploy_dlp_full_stack.yml index 7a7eb7c..fe25228 100644 --- a/ansible/deploy_dlp_full_stack.yml +++ b/ansible/deploy_dlp_full_stack.yml @@ -8,3 +8,5 @@ gather_facts: false roles: - role: dlp-policy-engine + - role: dlp-content-analysis + - role: dlp-integrations diff --git a/ansible/group_vars/all.example.yml b/ansible/group_vars/all.example.yml index bb87c41..d08d820 100644 --- a/ansible/group_vars/all.example.yml +++ b/ansible/group_vars/all.example.yml @@ -40,4 +40,6 @@ aw_dlp_policy_engine_enabled: true aw_dlp_policy_engine_bind_host: "0.0.0.0" aw_dlp_policy_engine_port: 5601 aw_dlp_policy_engine_db_path: "{{ aw_server_data_dir }}/dlp-policy-engine.sqlite" +aw_dlp_content_analysis_enabled: true +aw_dlp_integrations_enabled: true aw_server_post_deploy_health_check_enabled: true diff --git a/ansible/group_vars/all.yml b/ansible/group_vars/all.yml index 5b35b12..77de70a 100644 --- a/ansible/group_vars/all.yml +++ b/ansible/group_vars/all.yml @@ -31,6 +31,8 @@ aw_dlp_policy_engine_enabled: true aw_dlp_policy_engine_bind_host: "0.0.0.0" aw_dlp_policy_engine_port: 5601 aw_dlp_policy_engine_db_path: "{{ aw_server_data_dir }}/dlp-policy-engine.sqlite" +aw_dlp_content_analysis_enabled: true +aw_dlp_integrations_enabled: true aw_server_post_deploy_health_check_enabled: true aw_worktime_from: "08:00" diff --git a/ansible/roles/dlp-integrations/tasks/main.yml b/ansible/roles/dlp-integrations/tasks/main.yml new file mode 100644 index 0000000..9676b13 --- /dev/null +++ b/ansible/roles/dlp-integrations/tasks/main.yml @@ -0,0 +1,77 @@ +--- +- name: Ensure DLP integrations directory + ansible.builtin.file: + path: /opt/activitywatch/dlp-integrations + state: directory + owner: "{{ aw_server_user | default('activitywatch') }}" + group: "{{ aw_server_group | default('activitywatch') }}" + mode: "0755" + +- name: Copy DLP integrations files + ansible.builtin.copy: + src: "{{ playbook_dir }}/../aw-server/dlp-integrations/{{ item }}" + dest: "/opt/activitywatch/dlp-integrations/{{ item }}" + owner: "{{ aw_server_user | default('activitywatch') }}" + group: "{{ aw_server_group | default('activitywatch') }}" + mode: "0644" + loop: + - cef_exporter.py + - webhook_sender.py + - cef-config.yaml + - webhook-config.yaml + - requirements.txt + +- name: Make DLP integrations scripts executable + ansible.builtin.file: + path: "/opt/activitywatch/dlp-integrations/{{ item }}" + owner: "{{ aw_server_user | default('activitywatch') }}" + group: "{{ aw_server_group | default('activitywatch') }}" + mode: "0755" + loop: + - cef_exporter.py + - webhook_sender.py + +- name: Create venv for DLP integrations + ansible.builtin.command: + cmd: python3 -m venv /opt/activitywatch/dlp-integrations/.venv + args: + creates: /opt/activitywatch/dlp-integrations/.venv/bin/python + +- name: Install Python dependencies for DLP integrations + ansible.builtin.pip: + requirements: /opt/activitywatch/dlp-integrations/requirements.txt + virtualenv: /opt/activitywatch/dlp-integrations/.venv + +- name: Ensure integrations state directory + ansible.builtin.file: + path: /var/lib/activitywatch/dlp-integrations + state: directory + owner: "{{ aw_server_user | default('activitywatch') }}" + group: "{{ aw_server_group | default('activitywatch') }}" + mode: "0755" + +- name: Install systemd units/timers for DLP integrations + ansible.builtin.copy: + src: "{{ playbook_dir }}/../aw-server/dlp-integrations/{{ item.src }}" + dest: "{{ item.dest }}" + owner: root + group: root + mode: "0644" + 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: "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" } + +- name: Reload systemd + ansible.builtin.systemd: + daemon_reload: true + +- name: Enable and restart DLP integration timers + ansible.builtin.systemd: + name: "{{ item }}" + enabled: true + state: restarted + loop: + - aw-dlp-cef-exporter.timer + - aw-dlp-webhook-sender.timer diff --git a/aw-server/dlp-integrations/cef-config.yaml b/aw-server/dlp-integrations/cef-config.yaml new file mode 100644 index 0000000..40b038a --- /dev/null +++ b/aw-server/dlp-integrations/cef-config.yaml @@ -0,0 +1,10 @@ +aw_api_base: "http://127.0.0.1:5600/api/0" +state_path: "/var/lib/activitywatch/dlp-integrations/cef-state.json" +syslog_host: "127.0.0.1" +syslog_port: 514 +syslog_proto: "udp" +per_bucket_limit: 300 +severity_mapping: + low: 3 + medium: 6 + high: 10 diff --git a/aw-server/dlp-integrations/cef-exporter.service b/aw-server/dlp-integrations/cef-exporter.service index c5624a4..41186a0 100644 --- a/aw-server/dlp-integrations/cef-exporter.service +++ b/aw-server/dlp-integrations/cef-exporter.service @@ -6,7 +6,8 @@ After=network-online.target Type=oneshot User=activitywatch Group=activitywatch -ExecStart=/usr/bin/python3 /opt/activitywatch/dlp-integrations/cef_exporter.py +WorkingDirectory=/opt/activitywatch/dlp-integrations +ExecStart=/opt/activitywatch/dlp-integrations/.venv/bin/python /opt/activitywatch/dlp-integrations/cef_exporter.py [Install] WantedBy=multi-user.target diff --git a/aw-server/dlp-integrations/cef-exporter.timer b/aw-server/dlp-integrations/cef-exporter.timer index aaaad7e..4582564 100644 --- a/aw-server/dlp-integrations/cef-exporter.timer +++ b/aw-server/dlp-integrations/cef-exporter.timer @@ -4,7 +4,7 @@ Description=Run AWatch DLP CEF Exporter every 5 minutes [Timer] OnBootSec=2min OnUnitActiveSec=5min -Unit=cef-exporter.service +Unit=aw-dlp-cef-exporter.service Persistent=true [Install] diff --git a/aw-server/dlp-integrations/cef_exporter.py b/aw-server/dlp-integrations/cef_exporter.py index 64ac3ff..a21e966 100644 --- a/aw-server/dlp-integrations/cef_exporter.py +++ b/aw-server/dlp-integrations/cef_exporter.py @@ -3,23 +3,87 @@ from __future__ import annotations import json import logging -import os import socket from datetime import datetime, timezone +from pathlib import Path +from typing import Any +from urllib import error, request -logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") +import yaml + +LOG = logging.getLogger("aw.dlp.cef_exporter") -def build_cef(event: dict) -> str: - sev_map = {"low": 3, "medium": 6, "high": 10} - sev = sev_map.get(event.get("severity", "low"), 3) - ts = datetime.now(timezone.utc).isoformat() - msg = event.get("message", "") - host = event.get("hostname", "unknown") - return f"CEF:0|AWatch-rus|DLP|1.0|{event.get('id','dlp')}|{msg}|{sev}|rt={ts} shost={host}" +def setup_logging() -> None: + logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") -def send_syslog(line: str, host: str, port: int) -> None: +def load_yaml(path: Path) -> dict[str, Any]: + if not path.exists(): + return {} + data = yaml.safe_load(path.read_text(encoding="utf-8")) or {} + if not isinstance(data, dict): + return {} + return data + + +def load_json(path: Path) -> dict[str, Any]: + if not path.exists(): + return {} + try: + data = json.loads(path.read_text(encoding="utf-8")) + if isinstance(data, dict): + return data + except Exception: + return {} + return {} + + +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 escape_cef(v: Any) -> str: + s = "" if v is None else str(v) + return s.replace("\\", "\\\\").replace("|", "\\|").replace("=", "\\=").replace("\n", "\\n").replace("\r", "") + + +def map_severity(name: str, mapping: dict[str, int]) -> int: + return int(mapping.get((name or "").lower(), 3)) + + +def build_cef(event: dict[str, Any], mapping: dict[str, int]) -> str: + data = event.get("data") or {} + sev_name = str(data.get("severity") or "low").lower() + sev_num = map_severity(sev_name, mapping) + rt = event.get("timestamp") or datetime.now(timezone.utc).isoformat() + rule = data.get("ruleId") or "dlp-incident" + msg = data.get("message") or "AWatch DLP incident" + sig = data.get("signalType") or "unknown" + host = data.get("hostname") or "unknown" + user = data.get("username") or "unknown" + action = data.get("action") or "alert" + ext = ( + f"rt={escape_cef(rt)} " + f"shost={escape_cef(host)} " + f"suser={escape_cef(user)} " + f"cs1Label=signalType cs1={escape_cef(sig)} " + f"cs2Label=action cs2={escape_cef(action)} " + f"cs3Label=ruleId cs3={escape_cef(rule)}" + ) + return ( + f"CEF:0|AWatch-rus|DLP|1.0|{escape_cef(rule)}|{escape_cef(msg)}|{sev_num}|{ext}" + ) + + +def send_syslog_udp(line: str, host: str, port: int) -> None: sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) try: sock.sendto(line.encode("utf-8", errors="ignore"), (host, port)) @@ -27,14 +91,74 @@ def send_syslog(line: str, host: str, port: int) -> None: sock.close() +def send_syslog_tcp(line: str, host: str, port: int, timeout: int = 10) -> None: + sock = socket.create_connection((host, port), timeout=timeout) + try: + sock.sendall((line + "\n").encode("utf-8", errors="ignore")) + finally: + sock.close() + + +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 main() -> None: - sample = os.environ.get("AW_DLP_CEF_SAMPLE", "") - event = json.loads(sample) if sample else {"id": "startup", "message": "cef exporter heartbeat", "severity": "low"} - line = build_cef(event) - host = os.environ.get("AW_DLP_SYSLOG_HOST", "127.0.0.1") - port = int(os.environ.get("AW_DLP_SYSLOG_PORT", "514")) - send_syslog(line, host, port) - logging.info("sent CEF event to %s:%d", host, port) + setup_logging() + cfg_path = Path("/opt/activitywatch/dlp-integrations/cef-config.yaml") + cfg = load_yaml(cfg_path) + aw_base = str(cfg.get("aw_api_base", "http://127.0.0.1:5600/api/0")).rstrip("/") + 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")).lower() + per_bucket_limit = int(cfg.get("per_bucket_limit", 300)) + state_path = Path(str(cfg.get("state_path", "/var/lib/activitywatch/dlp-integrations/cef-state.json"))) + sev_mapping = cfg.get("severity_mapping", {"low": 3, "medium": 6, "high": 10}) + if not isinstance(sev_mapping, dict): + sev_mapping = {"low": 3, "medium": 6, "high": 10} + + 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 ev in incidents: + line = build_cef(ev, sev_mapping) + if syslog_proto == "tcp": + send_syslog_tcp(line, syslog_host, syslog_port) + else: + send_syslog_udp(line, syslog_host, syslog_port) + sent += 1 + state["last_ids"] = max_ids + state["updated_at"] = datetime.now(timezone.utc).isoformat() + save_json(state_path, state) + LOG.info("CEF exporter done: sent=%d buckets=%d target=%s:%d/%s", sent, len(max_ids), syslog_host, syslog_port, syslog_proto) if __name__ == "__main__": diff --git a/aw-server/dlp-integrations/requirements.txt b/aw-server/dlp-integrations/requirements.txt new file mode 100644 index 0000000..c1a201d --- /dev/null +++ b/aw-server/dlp-integrations/requirements.txt @@ -0,0 +1 @@ +PyYAML>=6.0 diff --git a/aw-server/dlp-integrations/webhook-config.yaml b/aw-server/dlp-integrations/webhook-config.yaml new file mode 100644 index 0000000..c0af9a1 --- /dev/null +++ b/aw-server/dlp-integrations/webhook-config.yaml @@ -0,0 +1,9 @@ +aw_api_base: "http://127.0.0.1:5600/api/0" +state_path: "/var/lib/activitywatch/dlp-integrations/webhook-state.json" +retries: 4 +timeout_sec: 15 +backoff_base: 2.0 +per_bucket_limit: 300 +critical_webhooks: + - url: "https://hooks.slack.com/services/REPLACE/ME" + severity: ["high"] diff --git a/aw-server/dlp-integrations/webhook-sender.service b/aw-server/dlp-integrations/webhook-sender.service new file mode 100644 index 0000000..425dd98 --- /dev/null +++ b/aw-server/dlp-integrations/webhook-sender.service @@ -0,0 +1,13 @@ +[Unit] +Description=AWatch DLP Webhook Sender +After=network-online.target + +[Service] +Type=oneshot +User=activitywatch +Group=activitywatch +WorkingDirectory=/opt/activitywatch/dlp-integrations +ExecStart=/opt/activitywatch/dlp-integrations/.venv/bin/python /opt/activitywatch/dlp-integrations/webhook_sender.py + +[Install] +WantedBy=multi-user.target diff --git a/aw-server/dlp-integrations/webhook-sender.timer b/aw-server/dlp-integrations/webhook-sender.timer new file mode 100644 index 0000000..29ed438 --- /dev/null +++ b/aw-server/dlp-integrations/webhook-sender.timer @@ -0,0 +1,11 @@ +[Unit] +Description=Run AWatch DLP Webhook sender every 2 minutes + +[Timer] +OnBootSec=90s +OnUnitActiveSec=2min +Unit=aw-dlp-webhook-sender.service +Persistent=true + +[Install] +WantedBy=timers.target diff --git a/aw-server/dlp-integrations/webhook_sender.py b/aw-server/dlp-integrations/webhook_sender.py index 4c2f3cc..49ddd6c 100644 --- a/aw-server/dlp-integrations/webhook_sender.py +++ b/aw-server/dlp-integrations/webhook_sender.py @@ -2,28 +2,153 @@ from __future__ import annotations import json -import os +import logging import time -from urllib import request +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.webhook_sender") -def post(url: str, payload: dict, retries: int = 3) -> bool: - body = json.dumps(payload).encode("utf-8") - for i in range(retries): +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 {} + if not isinstance(data, dict): + return {} + return data + + +def load_json(path: Path) -> dict[str, Any]: + if not path.exists(): + return {} + try: + data = json.loads(path.read_text(encoding="utf-8")) + if isinstance(data, dict): + return data + except Exception: + return {} + return {} + + +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 post_with_retry(url: str, payload: dict[str, Any], retries: int, timeout: int, backoff_base: float) -> bool: + body = json.dumps(payload, ensure_ascii=False).encode("utf-8") + headers = {"Content-Type": "application/json; charset=utf-8"} + for attempt in range(1, retries + 1): try: - req = request.Request(url, data=body, headers={"Content-Type": "application/json"}, method="POST") - with request.urlopen(req, timeout=10): - return True - except Exception: - time.sleep(2 ** i) + req = request.Request(url, data=body, headers=headers, method="POST") + with request.urlopen(req, timeout=timeout) as resp: + code = getattr(resp, "status", 200) + if 200 <= code < 300: + return True + except error.HTTPError as exc: + LOG.warning("webhook http error url=%s code=%s attempt=%d/%d", url, exc.code, attempt, retries) + except Exception as exc: + LOG.warning("webhook transport error url=%s err=%s attempt=%d/%d", url, exc, attempt, retries) + if attempt < retries: + time.sleep(backoff_base ** (attempt - 1)) return False +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: + events = http_json(f"{aw_base}/buckets/{bid}/events?limit={int(per_bucket_limit)}") + 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 should_send(severity: str, allowed: list[str]) -> bool: + return severity.lower() in {s.lower() for s in allowed} + + def main() -> None: - hooks = [h.strip() for h in os.environ.get("AW_DLP_CRITICAL_WEBHOOKS", "").split(",") if h.strip()] - payload = {"text": "AWatch DLP critical incident", "severity": "high"} - for hook in hooks: - post(hook, payload) + setup_logging() + cfg_path = Path("/opt/activitywatch/dlp-integrations/webhook-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/webhook-state.json"))) + retries = int(cfg.get("retries", 4)) + timeout = int(cfg.get("timeout_sec", 15)) + backoff_base = float(cfg.get("backoff_base", 2.0)) + per_bucket_limit = int(cfg.get("per_bucket_limit", 300)) + hooks = cfg.get("critical_webhooks", []) + if not isinstance(hooks, list): + hooks = [] + + 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 ev in incidents: + data = ev.get("data") or {} + severity = str(data.get("severity") or "low") + for hook in hooks: + if not isinstance(hook, dict): + continue + url = str(hook.get("url") or "").strip() + if not url: + continue + allowed = hook.get("severity", ["high"]) + if isinstance(allowed, str): + allowed = [allowed] + if not should_send(severity, [str(x) for x in allowed]): + continue + payload = { + "source": "AWatch-rus DLP", + "timestamp": ev.get("timestamp"), + "event_id": ev.get("id"), + "severity": severity, + "message": data.get("message"), + "ruleId": data.get("ruleId"), + "signalType": data.get("signalType"), + "hostname": data.get("hostname"), + "username": data.get("username"), + "action": data.get("action"), + "raw": data, + } + if post_with_retry(url=url, payload=payload, retries=retries, timeout=timeout, backoff_base=backoff_base): + sent += 1 + + state["last_ids"] = max_ids + state["updated_at"] = datetime.now(timezone.utc).isoformat() + save_json(state_path, state) + LOG.info("Webhook sender done: delivered=%d incidents_seen=%d", sent, len(incidents)) if __name__ == "__main__": diff --git a/docs/dlp-integrations.md b/docs/dlp-integrations.md new file mode 100644 index 0000000..01d3d72 --- /dev/null +++ b/docs/dlp-integrations.md @@ -0,0 +1,46 @@ +# DLP Integrations (SIEM/SOAR) + +## Что реализовано + +- `aw-server/dlp-integrations/cef_exporter.py` + Читает новые события из `aw-dlp-incidents_*`, конвертирует в CEF и отправляет в syslog. +- `aw-server/dlp-integrations/webhook_sender.py` + Читает новые события из `aw-dlp-incidents_*` и отправляет webhook для нужных severity. +- Systemd: + - `aw-dlp-cef-exporter.service` + `aw-dlp-cef-exporter.timer` (каждые 5 минут) + - `aw-dlp-webhook-sender.service` + `aw-dlp-webhook-sender.timer` (каждые 2 минуты) + +## Конфиги + +- `/opt/activitywatch/dlp-integrations/cef-config.yaml` +- `/opt/activitywatch/dlp-integrations/webhook-config.yaml` + +Поля: + +- `aw_api_base` — URL AW API (`http://127.0.0.1:5600/api/0`) +- `state_path` — файл состояния last processed event id +- `per_bucket_limit` — лимит чтения событий из каждого `aw-dlp-incidents_*` + +CEF: + +- `syslog_host`, `syslog_port`, `syslog_proto` (`udp`/`tcp`) +- `severity_mapping` (`low/medium/high` -> CEF severity) + +Webhook: + +- `critical_webhooks` список: + - `url` + - `severity` (например `["high"]`) +- `retries`, `timeout_sec`, `backoff_base` + +## Ansible + +Интегрировано в: + +- `ansible/deploy_aw_server.yml` +- `ansible/deploy_dlp_full_stack.yml` +- `ansible/roles/dlp-integrations/tasks/main.yml` + +Флаг включения: + +- `aw_dlp_integrations_enabled: true`