"""Inti Client Agent: registrasi, koneksi WebSocket, verifikasi perintah, penerapan kebijakan, pengawas, browser terkelola.""" from __future__ import annotations import asyncio import hashlib import hmac import json import logging import os import socket import sys import time from pathlib import Path import requests sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) import websockets # noqa: E402 from config import VERSION, Config, agent_home # noqa: E402 from modules import sysinfo # noqa: E402 from modules.browser_manager import BrowserController, Hooks, launch_args, profile_dir # noqa: E402 from modules.ipc import IpcServer # noqa: E402 from modules.policy_engine import PolicyEngine # noqa: E402 from modules.watcher import Watcher # noqa: E402 from modules.winops import IS_WINDOWS, Ops # noqa: E402 log = logging.getLogger("agent") HEARTBEAT_S = 5 def canonical(payload: dict) -> bytes: return json.dumps(payload, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode() def verify_command(cmd: dict, signing_key: str, seen_nonces: dict[str, float], now: float | None = None) -> tuple[bool, str]: """Perintah harus: bertanda tangan HMAC valid, belum kedaluwarsa, dan nonce belum pernah dipakai (anti-replay).""" now = now or time.time() try: payload = {k: cmd[k] for k in ("id", "action", "params", "nonce", "exp")} sig = str(cmd["sig"]) except KeyError: return False, "format perintah salah" want = hmac.new(signing_key.encode(), canonical(payload), hashlib.sha256).hexdigest() if not hmac.compare_digest(want, sig): return False, "tanda tangan tidak valid" if payload["exp"] < now: return False, "perintah kedaluwarsa" for n, t in list(seen_nonces.items()): if t < now - 900: seen_nonces.pop(n, None) if payload["nonce"] in seen_nonces: return False, "perintah diputar ulang (replay)" seen_nonces[payload["nonce"]] = now return True, "ok" class Agent: def __init__(self, cfg: Config, ops: Ops | None = None): self.cfg = cfg self.ops = ops or Ops(cfg.dry_run) self.policy: dict = {} self.policy_version = "" self.ws = None self.nonces: dict[str, float] = {} self.lock_state = False self.server_ip = self._server_ip() self.engine = PolicyEngine(self.ops, self.server_ip, cfg.applocker, cfg.protect_task_manager) self.ipc = IpcServer(self._on_helper) self.helper_state = {"state": "idle", "user": "", "downloads_dir": ""} self.ports = {"chrome": cfg.chrome_port, "edge": cfg.edge_port} self.watcher = Watcher(self.ports, (profile_dir("chrome"), profile_dir("edge"))) self.controllers: dict[str, BrowserController] = {} self.pending_downloads: dict[int, dict] = {} self.pending_apps: set[int] = set() self._screen_on = False self._applying = asyncio.Lock() self._stop = asyncio.Event() self.permit_task: asyncio.Task | None = None # ---- util --------------------------------------------------------------------------------------------------------- def _server_ip(self) -> str | None: try: from urllib.parse import urlparse h = urlparse(self.cfg.server_url).hostname return socket.gethostbyname(h) if h else None except OSError: return None def http(self, method: str, path: str, body: dict | None = None, timeout: int = 15): r = requests.request(method, self.cfg.server_url.rstrip("/") + path, json=body, timeout=timeout, verify=self.cfg.verify, headers={"Authorization": f"Bearer {self.cfg.token}"}) return r async def ahttp(self, method: str, path: str, body: dict | None = None): try: return await asyncio.to_thread(self.http, method, path, body) except requests.RequestException as e: log.warning("HTTP %s %s gagal: %s", method, path, e) return None async def send(self, msg: dict) -> None: if self.ws is not None: try: await self.ws.send(json.dumps(msg)) except Exception as e: log.debug("send gagal: %s", e) async def toast(self, text: str, seconds: int = 8) -> None: await self.ipc.send({"cmd": "toast", "text": text, "seconds": seconds}) # ---- registrasi ---------------------------------------------------------------------------------------------------- @staticmethod def register(cfg: Config, enrollment_key: str) -> None: r = requests.post(cfg.server_url.rstrip("/") + "/api/agent/register", timeout=20, verify=cfg.verify, json={ "enrollment_key": enrollment_key, "device_uid": cfg.device_uid, "hostname": socket.gethostname(), "ip_address": sysinfo.local_ip(), "mac_address": sysinfo.mac_address(), "os_name": sysinfo.os_name(), "windows_user": sysinfo.active_user(), "agent_version": VERSION}) if r.status_code != 200: raise RuntimeError(f"Registrasi ditolak server ({r.status_code}): {r.text[:200]}") d = r.json() cfg.token, cfg.signing_key, cfg.computer_id, cfg.token_issued_at = d["token"], d["signing_key"], d["computer_id"], time.time() cfg.save() # ---- kebijakan ----------------------------------------------------------------------------------------------------------- def _policy_cache(self) -> Path: return agent_home() / "policy.json" def load_cached_policy(self) -> None: f = self._policy_cache() if f.exists(): try: self.policy = json.loads(f.read_text(encoding="utf-8")) self.policy_version = self.policy.get("version", "") except (OSError, json.JSONDecodeError): pass async def apply_policy(self, policy: dict, notify_server: bool = True) -> None: async with self._applying: self.policy, self.policy_version = policy, policy.get("version", "") try: # atomik: tulis ke berkas sementara lalu ganti, supaya cache tidak pernah setengah-tertulis f = self._policy_cache() tmp = f.with_suffix(".tmp") tmp.write_text(json.dumps(policy), encoding="utf-8") os.replace(tmp, f) except OSError: pass self.watcher.set_policy(policy) errors = await asyncio.to_thread(self.engine.apply, policy, self.watcher.permit_active) lock = policy.get("lock", {}) self.lock_state = bool(lock.get("locked")) await self.ipc.send({"cmd": "lock" if self.lock_state else "unlock", "message": lock.get("message", "")}) for c in self.controllers.values(): asyncio.create_task(c.recheck_all()) if notify_server: await self.send({"type": "policy_applied", "version": self.policy_version, "errors": errors}) log.info("kebijakan %s diterapkan (%d galat)", self.policy_version, len(errors)) # ---- helper (sesi pengguna) ------------------------------------------------------------------------------------------------- async def _on_helper(self, m: dict) -> None: t = m.get("type") if t == "hello": self.helper_state["downloads_dir"] = m.get("downloads_dir", "") self.helper_state["user"] = m.get("user", "") lock = self.policy.get("lock", {}) if lock.get("locked"): await self.ipc.send({"cmd": "lock", "message": lock.get("message", "")}) elif t == "state": self.helper_state.update(state=m.get("state", "idle"), user=m.get("user", "")) elif t == "frame": await self.send({"type": "screen_frame", "jpeg": m.get("jpeg"), "w": m.get("w"), "h": m.get("h")}) elif t == "control_state": await self.send({"type": "control_state", "state": m.get("state")}) elif t == "helper_gone": self._screen_on = False def user_downloads(self) -> Path: d = self.helper_state.get("downloads_dir") return Path(d) if d else Path.home() / "Downloads" # ---- hook browser ------------------------------------------------------------------------------------------------------- def make_hooks(self) -> Hooks: async def blocked_site(kind, domain, reason): await self.event("blocked_site", {"domain": domain, "browser": kind, "reason": reason}) await self.toast(f"Situs diblokir: {domain}", 4) async def tab_limit(kind, limit): await self.event("tab_limit", {"browser": kind, "limit": limit}) await self.toast(f"Batas tab ({limit}) tercapai. Tutup tab lain terlebih dahulu.", 5) async def dl_blocked(kind, url, name): await self.request_download(kind, url, name) async def dl_used(rid): await self.ahttp("POST", f"/api/agent/downloads/{rid}/used") return Hooks(blocked_site, tab_limit, dl_blocked, dl_used, self.toast) async def request_download(self, kind: str, url: str, name: str) -> None: from modules.browser_manager import safe_filename fname = safe_filename(name or url.rsplit("/", 1)[-1].split("?")[0] or "unduhan") size, ctype = await asyncio.to_thread(self._probe, url) r = await self.ahttp("POST", "/api/agent/downloads", {"file_name": fname, "file_size": size, "file_type": ctype, "source_url": url[:2000], "browser": kind}) if r is None or r.status_code != 200: await self.toast("Permintaan download gagal dikirim ke guru. Coba lagi.", 6) return d = r.json() self.pending_downloads[d["id"]] = {"url": url, "name": fname, "kind": kind} if d["status"] == "ditolak": await self.toast(f"Download ditolak: {d.get('decision_note') or fname}", 8) self.pending_downloads.pop(d["id"], None) elif d["status"] == "disetujui": await self._download_decision({"request_id": d["id"], "status": "disetujui", "valid_minutes": 5}) else: await self.toast(f"Download '{fname}' menunggu persetujuan guru…", 8) @staticmethod def _probe(url: str) -> tuple[int | None, str | None]: if not url.startswith(("http://", "https://")): return None, None try: r = requests.head(url, timeout=3, allow_redirects=True) n = r.headers.get("Content-Length") return (int(n) if n and n.isdigit() else None), (r.headers.get("Content-Type") or "")[:120] or None except requests.RequestException: return None, None async def _download_decision(self, m: dict) -> None: rid = m["request_id"] info = self.pending_downloads.get(rid) if m["status"] == "disetujui" and info: ctrl = self.controllers.get(info["kind"]) or next(iter(self.controllers.values()), None) if ctrl: await ctrl.add_approval(rid, info["url"], info["name"], int(m.get("valid_minutes") or 5)) await self.toast(f"Download '{info['name']}' DISETUJUI. Silakan klik unduh lagi dalam {m.get('valid_minutes') or 5} menit.", 12) elif m["status"] in ("ditolak", "kedaluwarsa"): await self.toast(f"Download '{(info or {}).get('name', m.get('file_name', ''))}' {m['status'].upper()}" + (f": {m['note']}" if m.get("note") else ""), 10) self.pending_downloads.pop(rid, None) # ---- event ke server ------------------------------------------------------------------------------------------------------ async def event(self, type_: str, detail: dict) -> None: await self.ahttp("POST", "/api/agent/events", {"type": type_, "detail": detail}) # ---- perintah ------------------------------------------------------------------------------------------------------------- async def handle_command(self, cmd: dict) -> None: ok, why = verify_command(cmd, self.cfg.signing_key, self.nonces) if not ok: log.warning("perintah ditolak: %s", why) await self.event("tamper", {"message": f"Perintah ditolak: {why}", "action": str(cmd.get("action"))[:30]}) await self.send({"type": "command_result", "id": cmd.get("id", 0), "ok": False, "detail": f"Ditolak agent: {why}"}) return action, p, cid = cmd["action"], cmd.get("params") or {}, cmd["id"] log.info("perintah diterima: %s (id=%s)", action, cid) try: ok, detail = await self.execute(action, p) except Exception as e: log.exception("perintah %s gagal", action) ok, detail = False, f"{type(e).__name__}: {e}" log.info("perintah %s selesai: %s - %s", action, "OK" if ok else "GAGAL", detail) if cid: # id 0 = kendali layar internal, tanpa pelaporan await self.send({"type": "command_result", "id": cid, "ok": ok, "detail": detail}) async def execute(self, action: str, p: dict) -> tuple[bool, str]: run = lambda a: asyncio.to_thread(self.ops.run, a) if action in ("shutdown", "restart"): delay = int(p.get("delay", 30)) flag = "/s" if action == "shutdown" else "/r" await self.toast(f"Komputer akan {'dimatikan' if action == 'shutdown' else 'dimulai ulang'} dalam {delay} detik oleh guru.", delay) rc, out = await run(["shutdown", flag, "/t", str(delay), "/c", "Perintah dari guru (Classroom Management System)", "/d", "p:4:1"]) return rc == 0, out or f"shutdown {flag} dijadwalkan ({delay} dtk)" if action == "abort_shutdown": rc, out = await run(["shutdown", "/a"]) return rc == 0, out or "dibatalkan" if action == "logout": return (await self._helper_cmd({"cmd": "logout"}, "logout dikirim ke sesi pengguna")) if action == "lock_workstation": return (await self._helper_cmd({"cmd": "lock_workstation"}, "Windows dikunci")) if action == "lock": self.lock_state = True return (await self._helper_cmd({"cmd": "lock", "message": "Komputer dikunci oleh guru"}, "layar dikunci")) if action == "unlock": self.lock_state = False return (await self._helper_cmd({"cmd": "unlock"}, "kunci dibuka")) if action == "message": return (await self._helper_cmd({"cmd": "message", "text": p["text"], "seconds": p.get("seconds", 15), "from": p.get("from", "Guru")}, "pesan ditampilkan")) if action == "refresh_policy": r = await self.ahttp("GET", "/api/agent/policy") if r is None or r.status_code != 200: return False, "gagal mengambil kebijakan" await self.apply_policy(r.json()) return True, "kebijakan diperbarui" if action == "close_browser": n = 0 for c in self.controllers.values(): n += 1 if await c.close_all() else 0 killed = 0 for exe in ("chrome.exe", "msedge.exe"): killed += await asyncio.to_thread(Watcher.kill_by_name, exe) if not self.ops.dry_run else 0 return True, f"browser ditutup ({n} terkelola, {killed} proses)" if action == "kill_process": if self.ops.dry_run: return True, f"[dry-run] akan menutup {p['name']}" n = await asyncio.to_thread(Watcher.kill_by_name, p["name"]) return n > 0, f"{n} proses {p['name']} dihentikan" if n else "proses tidak ditemukan / dilindungi" if action == "release": asyncio.create_task(self._release()) return True, "perangkat dilepas, penegakan dibersihkan" if action == "screen_start": self._screen_on = True return (await self._helper_cmd({"cmd": "screen_start", "viewer": p.get("viewer", "Guru"), "fps": p.get("fps", 2), "quality": p.get("quality", 50), "max_width": p.get("max_width", 1280)}, "pemantauan layar dimulai")) if action == "screen_stop": self._screen_on = False return (await self._helper_cmd({"cmd": "screen_stop"}, "pemantauan layar dihentikan")) if action in ("control_request", "control_stop"): return (await self._helper_cmd({"cmd": "control_on" if action == "control_request" else "control_off", "viewer": p.get("viewer", "Guru")}, "ok")) if action == "control_input": return (await self._helper_cmd({"cmd": "control_input", "event": p.get("event") or {}}, "ok")) return False, f"aksi tidak dikenal: {action}" async def _helper_cmd(self, msg: dict, ok_text: str) -> tuple[bool, str]: if await self.ipc.send(msg): return True, ok_text return False, "helper sesi pengguna tidak aktif (tidak ada pengguna login?)" async def _release(self) -> None: await asyncio.sleep(1) await asyncio.to_thread(self.engine.clear_all) try: self.cfg.path.unlink(missing_ok=True) self._policy_cache().unlink(missing_ok=True) except OSError: pass self._stop.set() if self.ws is not None: # putuskan koneksi supaya loop utama keluar & service berhenti sendiri try: await self.ws.close() except Exception: pass # ---- pengawas proses ------------------------------------------------------------------------------------------------------------- async def watch_loop(self) -> None: while not self._stop.is_set(): await asyncio.sleep(2) try: actions = await asyncio.to_thread(self.watcher.scan) for a in actions: await self.handle_watch_action(a) except Exception: log.exception("watch_loop") async def handle_watch_action(self, a) -> None: key = f"{a.kind}:{a.name}" if a.kind == "launch_managed": if not self.watcher.should_report(key, 5): return await asyncio.to_thread(Watcher.kill, a.pid) kind = a.detail args = launch_args(kind, self.ports[kind]) if not self.ops.dry_run else [kind] hp = self.policy.get("browser", {}).get("homepage") await self.ipc.send({"cmd": "launch_browser", "args": args + ([hp] if hp and hp.startswith("http") else [])}) await self.toast("Gunakan browser sekolah yang dikelola. Browser dibuka ulang otomatis.", 5) return await asyncio.to_thread(Watcher.kill, a.pid) if not self.watcher.should_report(key, 60): return if a.kind == "kill": await self.event("blocked_app", {"app": a.name, "reason": a.reason}) await self.toast(f"Aplikasi diblokir: {a.name} ({a.reason})", 6) else: # request_install | request_store r = await self.ahttp("POST", "/api/agent/app-requests", {"app_name": a.detail or a.name, "request_type": "store" if a.kind == "request_store" else "install", "detail": a.reason}) if r is not None and r.status_code == 200: self.pending_apps.add(r.json()["id"]) await self.toast(f"Instalasi '{a.detail or a.name}' diblokir. Permintaan dikirim ke guru.", 8) async def _permit(self, m: dict) -> None: if m["status"] == "disetujui": minutes = int(m.get("minutes") or 15) self.watcher.grant_permit(minutes) await self.apply_policy(self.policy, notify_server=False) await self.toast(f"Instalasi '{m.get('app_name', '')}' DISETUJUI selama {minutes} menit. Silakan jalankan installer.", 15) if self.permit_task: self.permit_task.cancel() self.permit_task = asyncio.create_task(self._permit_expiry(minutes)) else: await self.toast(f"Instalasi '{m.get('app_name', '')}' DITOLAK" + (f": {m['note']}" if m.get("note") else ""), 10) async def _permit_expiry(self, minutes: int) -> None: await asyncio.sleep(minutes * 60 + 1) await self.apply_policy(self.policy, notify_server=False) await self.toast("Izin instalasi berakhir.", 5) # ---- loop pendukung --------------------------------------------------------------------------------------------------------------------- def heartbeat_payload(self) -> dict: br = [i for c in self.controllers.values() if (i := c.info())] state = "locked" if self.lock_state else self.helper_state.get("state", "idle") return {"type": "heartbeat", "state": state, "user": self.helper_state.get("user") or sysinfo.active_user(), "ip": sysinfo.local_ip(self.server_ip or "8.8.8.8"), "mac": sysinfo.mac_address(), "agent_version": VERSION, "metrics": sysinfo.metrics(), "processes": sysinfo.running_apps(), "browsers": br, "policy_version": self.policy_version, "dry_run": self.ops.dry_run} async def heartbeat_loop(self) -> None: while not self._stop.is_set(): await asyncio.sleep(HEARTBEAT_S) await self.send(self.heartbeat_payload()) async def inventory_loop(self) -> None: await asyncio.sleep(10) while not self._stop.is_set(): apps = await asyncio.to_thread(sysinfo.installed_applications) if apps: await self.ahttp("POST", "/api/agent/inventory", {"applications": apps}) await asyncio.sleep(1800) async def poll_loop(self) -> None: """Cadangan bila pesan WebSocket terlewat: cek status permintaan yang masih menunggu.""" while not self._stop.is_set(): await asyncio.sleep(15) for rid in list(self.pending_downloads): r = await self.ahttp("GET", f"/api/agent/downloads/{rid}") if r is not None and r.status_code == 200 and r.json()["status"] != "menunggu": d = r.json() if d["status"] == "disetujui" and d.get("valid_until") and not d.get("used"): await self._download_decision({"request_id": rid, "status": "disetujui", "valid_minutes": 5}) elif d["status"] in ("ditolak", "kedaluwarsa"): await self._download_decision({"request_id": rid, "status": d["status"], "note": d.get("decision_note")}) if d["status"] != "menunggu": self.pending_downloads.pop(rid, None) if d["status"] != "disetujui" else None async def token_loop(self) -> None: while not self._stop.is_set(): await asyncio.sleep(3600) if time.time() - self.cfg.token_issued_at > 7 * 86400: r = await self.ahttp("POST", "/api/agent/rotate") if r is not None and r.status_code == 200: self.cfg.token, self.cfg.token_issued_at = r.json()["token"], time.time() self.cfg.save() log.info("token perangkat dirotasi") async def browser_tick_loop(self) -> None: while not self._stop.is_set(): await asyncio.sleep(2) for c in self.controllers.values(): await c.tick() # ---- koneksi ----------------------------------------------------------------------------------------------------------------------------------- async def ws_session(self) -> None: import ssl ssl_ctx = None if self.cfg.ws_url.startswith("wss://"): ssl_ctx = ssl.create_default_context(cafile=self.cfg.ca_file) if self.cfg.ca_file and Path(self.cfg.ca_file).exists() else ssl.create_default_context() hdr = {"Authorization": f"Bearer {self.cfg.token}"} try: conn = websockets.connect(self.cfg.ws_url, additional_headers=hdr, ssl=ssl_ctx, max_size=8_000_000, ping_interval=20, ping_timeout=20) except TypeError: conn = websockets.connect(self.cfg.ws_url, extra_headers=hdr, ssl=ssl_ctx, max_size=8_000_000, ping_interval=20, ping_timeout=20) async with conn as ws: self.ws = ws log.info("tersambung ke server") async for raw in ws: m = json.loads(raw) t = m.get("type") if t == "policy_update": await self.apply_policy(m["policy"]) elif t == "command": asyncio.create_task(self.handle_command(m)) elif t == "download_decision": await self._download_decision(m) elif t == "install_permit": await self._permit(m) async def run(self) -> None: self.load_cached_policy() await self.ipc.start() if self.policy: # penegakan tetap berlaku walaupun server mati / belum tersambung self.watcher.set_policy(self.policy) await self.apply_policy(self.policy, notify_server=False) hooks = self.make_hooks() for kind, port in self.ports.items(): ctrl = BrowserController(kind, port, hooks, self.user_downloads, agent_home() / "quarantine" / kind, lambda: self.policy) self.controllers[kind] = ctrl asyncio.create_task(ctrl.run()) loops = [self.watch_loop(), self.heartbeat_loop(), self.inventory_loop(), self.poll_loop(), self.token_loop(), self.browser_tick_loop()] tasks = [asyncio.create_task(c) for c in loops] backoff = 2 while not self._stop.is_set(): try: await self.ws_session() backoff = 2 except websockets.exceptions.InvalidStatus as e: code = getattr(getattr(e, "response", None), "status_code", 0) log.error("server menolak koneksi (%s)", code) if code in (401, 403): log.error("token tidak valid: perangkat mungkin dilepas. Daftarkan ulang agent.") except Exception as e: log.warning("koneksi terputus: %s", e) self.ws = None try: await asyncio.wait_for(self._stop.wait(), backoff) except asyncio.TimeoutError: pass backoff = min(backoff * 2, 30) for t in tasks: t.cancel() for c in self.controllers.values(): c.stop() await self.ipc.stop()