"""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: ... @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) pool.disconnect() except Exception as exc: raise MikrotikProvisioningError(f"RouterOS API gagal membuat Hotspot User: {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 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): resource = api.get_resource('/ip/hotspot/user') existing = resource.get(name=username) values = {"password": password, "profile": profile or "default"} 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)