"""Menerjemahkan kebijakan efektif dari server menjadi penegakan nyata di Windows. Lapisan penegakan: 1. Windows Firewall -> blokir seluruh internet (mode block_all), kecuali LAN & server guru. 2. Berkas hosts -> blokir domain terdaftar untuk semua aplikasi (bukan hanya browser). 3. Kebijakan Chrome/Edge (registry HKLM\\SOFTWARE\\Policies) -> URLBlocklist/URLAllowlist, matikan DoH, incognito, DevTools, ekstensi. 4. Microsoft Store / AppLocker -> kebijakan Windows resmi (Store: Enterprise/Education; AppLocker: Enterprise/Education). Penegakan lain (batas tab, persetujuan download, blokir installer) dilakukan modul browser_manager & watcher. """ from __future__ import annotations import ipaddress import logging import re from urllib.parse import urlparse from xml.sax.saxutils import quoteattr from .winops import Ops log = logging.getLogger("agent.policy") FW_GROUP = "CMS-Agent" HOSTS = r"C:\Windows\System32\drivers\etc\hosts" BEGIN, END = "# BEGIN CMS-AGENT (jangan diubah manual)", "# END CMS-AGENT" CHROME = r"SOFTWARE\Policies\Google\Chrome" EDGE = r"SOFTWARE\Policies\Microsoft\Edge" STORE = r"SOFTWARE\Policies\Microsoft\WindowsStore" ALWAYS_ALLOWED_V4 = ["127.0.0.0/8", "10.0.0.0/8", "172.16.0.0/12", "192.168.0.0/16", "169.254.0.0/16", "224.0.0.0/4", "255.255.255.255/32"] def blocked_ranges_v4(server_ip: str | None) -> list[str]: """Seluruh IPv4 dikurangi LAN & server guru -> daftar CIDR yang diblokir firewall.""" nets = [ipaddress.ip_network("0.0.0.0/0")] for a in ALWAYS_ALLOWED_V4 + ([f"{server_ip}/32"] if server_ip else []): ex = ipaddress.ip_network(a) new = [] for n in nets: if n.overlaps(ex): if n.subnet_of(ex): continue new.extend(n.address_exclude(ex)) else: new.append(n) nets = new return [str(n) for n in ipaddress.collapse_addresses(nets)] def host_of(url: str) -> str: try: return (urlparse(url).hostname or "").lower().rstrip(".") except ValueError: return "" def match_len(host: str, domains: list[str]) -> int: """Panjang domain terpanjang yang cocok dengan host (suffix, batas titik); 0 jika tidak ada.""" best = 0 for d in domains: if host == d or host.endswith("." + d): best = max(best, len(d)) return best def url_verdict(url: str, internet: dict) -> tuple[bool, str]: """(diizinkan?, alasan). Aturan 'paling spesifik menang'. Seri pada mode blocklist -> izinkan (whitelist menang, sama seperti server); pada mode allowlist, domain harus lebih spesifik dari aturan blokir agar lolos.""" u = urlparse(url) if u.scheme in ("about", "chrome", "edge", "devtools", "chrome-extension", "chrome-error", "chrome-untrusted", "data", "blob", "file", ""): return True, "internal" if u.scheme not in ("http", "https"): return True, "non-web" host = host_of(url) if host in ("localhost", "127.0.0.1", "::1"): return True, "localhost" mode = internet.get("mode", "allow_all") if mode == "allow_all": return True, "allow_all" if mode == "block_all": return False, "internet diblokir" a, b = match_len(host, internet.get("allow_domains", [])), match_len(host, internet.get("block_domains", [])) if mode == "allowlist": return (a > 0 and a > b), ("tidak ada di daftar izin" if not a else "diblokir") # blocklist (VPN/proxy selalu masuk block_domains) if b and b > a: return False, "situs diblokir" return True, "ok" def domains_to_url_patterns(domains: list[str]) -> list[str]: return sorted(set(domains)) class PolicyEngine: def __init__(self, ops: Ops, server_ip: str | None = None, applocker: bool = False, protect_task_manager: bool = False): self.ops, self.server_ip, self.applocker, self.protect_tm = ops, server_ip, applocker, protect_task_manager # -- pintu masuk ------------------------------------------------------------------------------------ def apply(self, policy: dict, install_permit_active: bool = False) -> list[str]: errors: list[str] = [] for name, fn in (("firewall", lambda: self.firewall(policy["internet"])), ("hosts", lambda: self.hosts(policy["internet"])), ("browser", lambda: self.browser_policies(policy["internet"], policy["browser"])), ("aplikasi", lambda: self.apps(policy["applications"], install_permit_active))): try: fn() except Exception as e: # satu lapisan gagal tidak menghentikan lapisan lain log.exception("lapisan %s gagal", name) errors.append(f"{name}: {e}") return errors def clear_all(self) -> None: """Dipakai saat perangkat dilepas / agent dicopot: hapus semua penegakan.""" self.ops.powershell(f"Remove-NetFirewallRule -Group '{FW_GROUP}' -ErrorAction SilentlyContinue") self._write_hosts([]) for k in (CHROME, EDGE, STORE): self.ops.reg_delete_tree(k) self.ops.reg_delete_value(r"SOFTWARE\Microsoft\Windows\CurrentVersion\Policies\System", "DisableTaskMgr") if self.applocker: self.ops.powershell("Set-AppLockerPolicy -XmlPolicy (Join-Path $env:TEMP 'cms-empty.xml') -ErrorAction SilentlyContinue") # -- firewall ------------------------------------------------------------------------------------------ def firewall(self, internet: dict) -> None: ps = [f"Remove-NetFirewallRule -Group '{FW_GROUP}' -ErrorAction SilentlyContinue"] if internet.get("mode") == "block_all": v4 = blocked_ranges_v4(self.server_ip) # Windows Firewall menolak daftar sangat panjang dalam satu aturan -> pecah per 40 CIDR for i in range(0, len(v4), 40): chunk = ",".join(v4[i:i + 40]) ps.append(f"New-NetFirewallRule -DisplayName 'CMS Blokir Internet v4 #{i // 40 + 1}' -Group '{FW_GROUP}' -Direction Outbound " f"-Action Block -Profile Any -Enabled True -RemoteAddress {chunk}") ps.append(f"New-NetFirewallRule -DisplayName 'CMS Blokir Internet v6' -Group '{FW_GROUP}' -Direction Outbound -Action Block " f"-Profile Any -Enabled True -RemoteAddress 2000::/3") rc, out = self.ops.powershell("; ".join(ps)) if rc != 0: raise RuntimeError(f"firewall rc={rc}: {out[:200]}") # -- hosts ----------------------------------------------------------------------------------------------- def hosts(self, internet: dict) -> None: doms = internet.get("block_domains", []) if internet.get("mode") in ("blocklist", "allowlist", "block_all") else [] self._write_hosts(doms) def _write_hosts(self, domains: list[str]) -> None: cur = self.ops.read_text(HOSTS) cur = re.sub(rf"\r?\n?{re.escape(BEGIN)}.*?{re.escape(END)}\r?\n?", "\n", cur, flags=re.S).rstrip() + "\n" if domains: lines = [BEGIN] for d in domains[:2500]: lines.append(f"0.0.0.0 {d}") if not d.startswith("www."): lines.append(f"0.0.0.0 www.{d}") lines.append(END) cur += "\n" + "\n".join(lines) + "\n" self.ops.write_text(HOSTS, cur) self.ops.run(["ipconfig", "/flushdns"]) # -- Chrome / Edge ------------------------------------------------------------------------------------------ def browser_policies(self, internet: dict, browser: dict) -> None: mode = internet.get("mode", "allow_all") for root, incog in ((CHROME, "IncognitoModeAvailability"), (EDGE, "InPrivateModeAvailability")): self.ops.reg_delete_tree(root) o = self.ops if mode == "block_all": block, allow = ["*"], [] elif mode == "allowlist": block, allow = ["*"], internet.get("allow_domains", []) elif mode == "blocklist": block, allow = internet.get("block_domains", []), internet.get("allow_domains", []) else: block, allow = [], [] if block: o.reg_set_list(root + r"\URLBlocklist", block) if allow: o.reg_set_list(root + r"\URLAllowlist", allow) o.reg_set(root, "DnsOverHttpsMode", "off", "sz") # DoH dimatikan supaya hosts/DNS tidak dilewati o.reg_set(root, "BuiltInDnsClientEnabled", 0) o.reg_set(root, incog, 1 if browser.get("block_incognito", True) else 0) o.reg_set(root, "BrowserGuestModeEnabled", 0) o.reg_set(root, "BrowserAddProfileEnabled", 0) o.reg_set(root, "DeveloperToolsAvailability", 2) # F12/DevTools dimatikan o.reg_set(root, "BlockExternalExtensions", 1) o.reg_set_list(root + r"\ExtensionInstallBlocklist", ["*"]) # cegah ekstensi VPN/proxy o.reg_set(root, "PasswordManagerEnabled", 0) hp = browser.get("homepage") or "" if hp: o.reg_set(root, "HomepageLocation", hp, "sz") o.reg_set(root, "HomepageIsNewTabPage", 0) o.reg_set(root, "ShowHomeButton", 1) o.reg_set(root, "RestoreOnStartup", 4) o.reg_set_list(root + r"\RestoreOnStartupURLs", [hp]) # -- Aplikasi / Store / AppLocker ------------------------------------------------------------------------------ def apps(self, apps: dict, permit_active: bool) -> None: self.ops.reg_delete_tree(STORE) if apps.get("block_store"): self.ops.reg_set(STORE, "RemoveWindowsStore", 1) # didukung Windows Enterprise/Education; Pro: dibantu watcher self.ops.reg_set(STORE, "DisableStoreApps", 1) tm = r"SOFTWARE\Microsoft\Windows\CurrentVersion\Policies\System" if self.protect_tm: self.ops.reg_set(tm, "DisableTaskMgr", 1) else: self.ops.reg_delete_value(tm, "DisableTaskMgr") if self.applocker: xml = build_applocker_xml(apps, permit_active) path = r"C:\ProgramData\ClassroomAgent\applocker.xml" self.ops.write_text(path, xml) self.ops.run(["sc.exe", "config", "AppIDSvc", "start=", "auto"]) self.ops.run(["sc.exe", "start", "AppIDSvc"]) rc, out = self.ops.powershell(f"Set-AppLockerPolicy -XmlPolicy '{path}'") if rc != 0: raise RuntimeError(f"AppLocker rc={rc}: {out[:200]}") def build_applocker_xml(apps: dict, permit_active: bool) -> str: """Kebijakan AppLocker: izinkan semua (default) + tolak aplikasi dilarang + (opsional) tolak installer di folder pengguna.""" allow_all = ('' '' '') rules, n = [], 10 blocked = list(apps.get("blocked_apps", [])) for exe in blocked: n += 1 rules.append(f'') if apps.get("block_installers") and not permit_active: for folder in ("Downloads", "Desktop", "Documents"): for ext in ("exe", "msi"): n += 1 rules.append(f'') exe_rules = allow_all + "".join(rules) return (f'{exe_rules}' f'' f'')