"""Explicit MikroTik provisioning boundary. No connection is made unless a caller supplies an adapter/API object. The mock is useful for tests and local simulations and is deliberately in-memory. """ from dataclasses import dataclass, field from typing import Protocol from routeros_api import RouterOsApiPool from app.core.secrets import decrypt_secret class MikrotikProvisioningError(RuntimeError): pass def _resource_id(item: dict) -> str: """Accept RouterOS API id formats returned by different client versions.""" resource_id = item.get(".id") or item.get("id") if not resource_id: raise MikrotikProvisioningError("RouterOS tidak mengembalikan ID resource") return resource_id def internet_probe_succeeded(replies: list[dict]) -> bool: """Interpret RouterOS /ping replies without assuming a fixed client version.""" for reply in replies or []: status = str(reply.get("status", "")).strip().lower() if status in {"", "echo reply", "reply", "ok"} and (reply.get("host") or reply.get("time") or not status): return True return False class MikrotikAdapter(Protocol): def upsert_pppoe_secret(self, username: str, password: str, profile: str | None = None) -> None: ... def set_pppoe_secret_enabled(self, username: str, enabled: bool) -> None: ... def delete_pppoe_secret(self, username: str) -> None: ... def add_hotspot_user(self, username: str, password: str, profile: str | None = None, uptime: str | None = None, idle: str | None = None, comment: str | None = None) -> None: ... def upsert_hotspot_profile(self, name: str, rate_limit: str, session_timeout: str, idle_timeout: str, shared: int) -> None: ... def delete_hotspot_profile(self, name: str) -> bool: ... @dataclass class RealMikrotikAdapter: """Real RouterOS API adapter with credentials injected from vault.""" host: str username: str password: str api_port: int = 8728 use_ssl: bool = False def _get_api(self): return RouterOsApiPool( host=self.host, username=self.username, password=self.password, port=self.api_port, use_ssl=self.use_ssl, plaintext_login=True, ) def upsert_pppoe_secret(self, username: str, password: str, profile: str | None = None) -> None: try: pool = self._get_api() api = pool.get_api() push_pppoe_user(api, username, password, profile) pool.disconnect() except Exception as exc: raise MikrotikProvisioningError(f"RouterOS API gagal membuat secret: {exc}") from exc def resolve_profile(self, requested: str | None) -> tuple[str, bool]: """Resolve an application profile against RouterOS without mutating it. Returns (effective_profile, used_default_fallback). The fallback is explicit so callers can audit a mismatch instead of sending an invalid RouterOS profile name. """ try: pool = self._get_api() try: api = pool.get_api() available = {str(row.get("name", "")) for row in api.get_resource("/ppp/profile").get()} finally: pool.disconnect() requested = (requested or "").strip() if requested and requested in available: return requested, False if "default" in available: return "default", bool(requested and requested != "default") raise MikrotikProvisioningError("Profile PPPoE default tidak tersedia di MikroTik") except MikrotikProvisioningError: raise except Exception as exc: raise MikrotikProvisioningError(f"RouterOS API gagal membaca profile PPPoE: {exc}") from exc def set_pppoe_secret_enabled(self, username: str, enabled: bool) -> None: try: pool = self._get_api() api = pool.get_api() if enabled: unsuspend_pppoe_user(api, username) else: suspend_pppoe_user(api, username) pool.disconnect() except Exception as exc: raise MikrotikProvisioningError(f"RouterOS API gagal update status secret: {exc}") from exc def add_hotspot_user(self, username: str, password: str, profile: str | None = None, uptime: str | None = None, idle: str | None = None, comment: str | None = None) -> None: try: pool = self._get_api() api = pool.get_api() push_hotspot_user(api, username, password, profile, limit_uptime=uptime, comment=comment) pool.disconnect() except Exception as exc: raise MikrotikProvisioningError(f"RouterOS API gagal membuat Hotspot User: {exc}") from exc def upsert_hotspot_profile(self, name: str, rate_limit: str, session_timeout: str, idle_timeout: str, shared: int) -> None: try: pool = self._get_api() api = pool.get_api() resource = api.get_resource('/ip/hotspot/user/profile') existing = resource.get(name=name) on_login_script = ( "# BW-VOUCHER-EXPIRY-BEGIN\n" ":local uid [/ip hotspot user find where name=$user];\n" ":if ([:len $uid] = 0) do={ :return; }\n" ":local c [/ip hotspot user get $uid comment];\n" ":if ([:pick $c 0 6] != \"bw:v1;\") do={ :return; }\n" ":local key \"validity=\";\n" ":local pos [:find $c $key];\n" ":if ([:typeof $pos] = \"nil\") do={ :return; }\n" ":local tail [:pick $c ($pos + [:len $key]) [:len $c]];\n" ":local stop [:find $tail \";\"];\n" ":if ([:typeof $stop] = \"nil\") do={ :set stop [:len $tail]; }\n" ":local sec [:tonum [:pick $tail 0 $stop]];\n" ":if ($sec < 60) do={ :return; }\n" ":local sn (\"bw-exp-\" . $user);\n" ":if ([:len [/system scheduler find where name=$sn]] = 0) do={\n" " :local ev (\":local u \\\"\" . $user . \"\\\"; /ip hotspot user remove [find where name=$u]; /ip hotspot active remove [find where user=$u]; /ip hotspot cookie remove [find where user=$u]; /system scheduler remove [find where name=\\\"\" . $sn . \"\\\"];\");\n" " /system scheduler add name=$sn interval=($sec . \"s\") start-time=startup on-event=$ev policy=read,write,policy,test comment=(\"bestweb:first-login:\" . $user);\n" "}\n" "# BW-VOUCHER-EXPIRY-END" ) values = { "rate-limit": rate_limit, "shared-users": str(shared), "session-timeout": session_timeout or "00:00:00", "idle-timeout": idle_timeout or "00:00:00", "on-login": on_login_script } if existing: resource.set(id=_resource_id(existing[0]), **values) else: resource.add(name=name, **values) pool.disconnect() except Exception as exc: raise MikrotikProvisioningError(f"RouterOS API gagal membuat Hotspot Profile: {exc}") from exc def delete_pppoe_secret(self, username: str) -> None: """Rollback helper for MikroTik.""" try: pool = self._get_api() api = pool.get_api() resource = api.get_resource('/ppp/secret') existing = resource.get(name=username) if existing: resource.remove(id=_resource_id(existing[0])) pool.disconnect() except Exception as exc: raise MikrotikProvisioningError(f"RouterOS API gagal menghapus secret: {exc}") from exc def delete_hotspot_profile(self, name: str) -> bool: """Remove a hotspot user profile from RouterOS.""" try: pool = self._get_api() api = pool.get_api() resource = api.get_resource('/ip/hotspot/user/profile') existing = resource.get(name=name) if existing: resource.remove(id=_resource_id(existing[0])) pool.disconnect() return True except MikrotikProvisioningError: raise except Exception as exc: raise MikrotikProvisioningError(f"RouterOS API gagal menghapus Hotspot Profile: {exc}") from exc def list_hotspot_users(self) -> list[dict]: """Read live hotspot user state (uptime, limit, disabled) from RouterOS.""" try: pool = self._get_api() try: api = pool.get_api() users = api.get_resource('/ip/hotspot/user').get() active = api.get_resource('/ip/hotspot/active').get() finally: pool.disconnect() active_users = {str(row.get("user") or "").strip() for row in active if row.get("user")} out = [] for row in users: name = str(row.get("name") or "").strip() if not name: continue out.append({ "name": name, "uptime": str(row.get("uptime") or ""), "limit_uptime": str(row.get("limit-uptime") or ""), "disabled": str(row.get("disabled") or "no") == "yes", "comment": str(row.get("comment") or ""), "profile": str(row.get("profile") or ""), "active": name in active_users, }) return out except Exception as exc: raise MikrotikProvisioningError(f"RouterOS API gagal membaca hotspot user: {exc}") from exc def delete_hotspot_user(self, username: str) -> bool: """Delete hotspot user and kick active session if present.""" try: pool = self._get_api() api = pool.get_api() resource = api.get_resource('/ip/hotspot/user') existing = resource.get(name=username) if existing: resource.remove(id=_resource_id(existing[0])) active_res = api.get_resource('/ip/hotspot/active') active_existing = active_res.get(user=username) if active_existing: active_res.remove(id=_resource_id(active_existing[0])) pool.disconnect() return True except Exception: return False def verify_pppoe(self, username: str) -> tuple[bool, bool]: """Read-only check: (secret_enabled, active_session).""" try: pool = self._get_api() try: api = pool.get_api() secret = api.get_resource('/ppp/secret').get(name=username) active = api.get_resource('/ppp/active').get(name=username) finally: pool.disconnect() if not secret: return False, bool(active) disabled = str(secret[0].get('disabled', '')).lower() in {'yes', 'true', '1'} return not disabled, bool(active) except Exception as exc: raise MikrotikProvisioningError(f"RouterOS API gagal memverifikasi PPPoE: {exc}") from exc def verify_pppoe_internet(self, username: str) -> tuple[bool, bool, bool]: """Read secret/session state and ping a public resolver from PPP client IP.""" try: pool = self._get_api() try: api = pool.get_api() secret = api.get_resource('/ppp/secret').get(name=username) active = api.get_resource('/ppp/active').get(name=username) if not secret: return False, bool(active), False disabled = str(secret[0].get('disabled', '')).lower() in {'yes', 'true', '1'} if disabled or not active: return not disabled, bool(active), False source = str(active[0].get('address', '')).strip() if not source: raise MikrotikProvisioningError("Sesi PPPoE tidak mengembalikan alamat IP pelanggan") replies = api.get_resource('/').call( 'ping', {'address': '1.1.1.1', 'count': '2', 'src-address': source} ) return True, True, internet_probe_succeeded(replies) finally: pool.disconnect() except MikrotikProvisioningError: raise except Exception as exc: raise MikrotikProvisioningError(f"RouterOS gagal probe internet PPPoE: {exc}") from exc @dataclass class MockMikrotikAdapter: """In-memory RouterOS simulation; no fabricated real-device success.""" secrets: dict[str, dict[str, str | bool]] = field(default_factory=dict) def upsert_pppoe_secret(self, username: str, password: str, profile: str | None = None) -> None: if not username: raise MikrotikProvisioningError("Username PPPoE belum diisi") self.secrets[username] = { "password": password, "profile": profile or "default", "service": "pppoe", "disabled": False, } def set_pppoe_secret_enabled(self, username: str, enabled: bool) -> None: if username not in self.secrets: raise MikrotikProvisioningError(f"Secret PPPoE {username} belum dibuat") self.secrets[username]["disabled"] = not enabled def add_hotspot_user(self, username: str, password: str, profile: str | None = None, uptime: str | None = None, idle: str | None = None, comment: str | None = None) -> None: pass def get_mikrotik_connection(config): """Create a real API pool only when explicitly called by an integration.""" return RouterOsApiPool(host=config.host, username=config.username, password=decrypt_secret(config.password), port=config.api_port or 8728, use_ssl=False, plaintext_login=True) def push_pppoe_user(api, username, password, profile): """Create/update a PPPoE secret on an explicitly supplied RouterOS API.""" resource = api.get_resource('/ppp/secret') existing = resource.get(name=username) values = {"password": password, "profile": profile or "default", "service": "pppoe", "disabled": "no"} if existing: resource.set(id=_resource_id(existing[0]), **values) else: resource.add(name=username, **values) def suspend_pppoe_user(api, username: str) -> None: resource = api.get_resource('/ppp/secret') existing = resource.get(name=username) if not existing: raise MikrotikProvisioningError(f"Secret PPPoE {username} belum dibuat") resource.set(id=_resource_id(existing[0]), disabled='yes') def unsuspend_pppoe_user(api, username: str) -> None: resource = api.get_resource('/ppp/secret') existing = resource.get(name=username) if not existing: raise MikrotikProvisioningError(f"Secret PPPoE {username} belum dibuat") resource.set(id=_resource_id(existing[0]), disabled='no') def push_hotspot_user(api, username, password, profile, limit_uptime=None, comment=None): resource = api.get_resource('/ip/hotspot/user') existing = resource.get(name=username) values = {"password": password, "profile": profile or "default"} if limit_uptime: values["limit-uptime"] = limit_uptime if comment: values["comment"] = comment if existing: resource.set(id=_resource_id(existing[0]), **values) else: resource.add(name=username, **values) def provision_customer_user(api, customer, password): if not customer.mikrotik_user: raise MikrotikProvisioningError('Username MikroTik belum diisi') # Every customer receives a PPPoE management secret, including hotspot # customers. Hotspot credentials remain a separate optional service. push_pppoe_user(api, customer.mikrotik_user, password, customer.service_profile) def push_hotspot_profile(api, name, rate_down, rate_up, session_timeout, shared): resource = api.get_resource('/ip/hotspot/user/profile') existing = resource.get(name=name) rate = f"{rate_up}k/{rate_down}k" values = { "rate-limit": rate, "shared-users": str(shared), "session-timeout": session_timeout or "00:00:00" } if existing: resource.set(id=_resource_id(existing[0]), **values) else: resource.add(name=name, **values)