from fastapi import APIRouter, Depends, HTTPException, Request, Form from fastapi.responses import HTMLResponse, RedirectResponse, JSONResponse from fastapi.templating import Jinja2Templates from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession import asyncio import os from app.core.database import get_db from app.core.security import require_roles from app.models.models import OLTDevice, MikrotikDevice, DeviceCredential from app.core.ssh_identity import test_pinned_ssh_identity templates = Jinja2Templates(directory="app/templates") router = APIRouter(dependencies=[Depends(require_roles("admin"))]) async def _probe_tcp(host: str, port: int): writer = None try: _, writer = await asyncio.open_connection(host, port) return True, None except OSError as exc: return False, str(exc) finally: if writer is not None: writer.close() await writer.wait_closed() def _json_probe(ok: bool, message: str): return JSONResponse({"ok": ok, "message": message}) @router.get("/devices", response_class=HTMLResponse) async def devices_page(request: Request, db: AsyncSession = Depends(get_db)): mikrotiks = (await db.execute(select(MikrotikDevice).order_by(MikrotikDevice.name))).scalars().all() olts = (await db.execute(select(OLTDevice).order_by(OLTDevice.name))).scalars().all() credentials = (await db.execute(select(DeviceCredential).order_by(DeviceCredential.name))).scalars().all() return templates.TemplateResponse( request=request, name="devices.html", context={ "mikrotiks": mikrotiks, "olts": olts, "mikrotik_credentials": [c for c in credentials if c.protocol == "mikrotik-api"], "olt_credentials": [c for c in credentials if c.protocol == "ssh"], "vault_error": False, }, ) @router.post("/devices/mikrotik") async def add_mikrotik( name: str = Form(...), host: str = Form(...), api_port: int = Form(8728), area: str = Form(""), db: AsyncSession = Depends(get_db), ): db.add(MikrotikDevice(name=name.strip(), host=host.strip(), api_port=api_port, area=area.strip())) await db.commit() return RedirectResponse("/devices", status_code=303) @router.post("/devices/olt") async def add_olt( name: str = Form(...), host: str = Form(...), vendor: str = Form("C-DATA"), model: str = Form("FD1601S-B1"), location: str = Form(""), pop_name: str = Form(""), management_protocol: str = Form("ssh"), api_port: int | None = Form(None), db: AsyncSession = Depends(get_db), ): db.add(OLTDevice(name=name.strip(), host=host.strip(), vendor=vendor.strip(), model=model.strip(), location=location.strip(), pop_name=pop_name.strip(), management_protocol=management_protocol.strip(), api_port=api_port or 22)) await db.commit() return RedirectResponse("/devices", status_code=303) @router.post("/devices/olt/{device_id}/update") async def update_olt(device_id: int, vendor: str = Form(...), model: str = Form(...), db: AsyncSession = Depends(get_db)): device = await db.get(OLTDevice, device_id) if not device: raise HTTPException(404, "OLT tidak ditemukan") device.vendor, device.model = vendor.strip(), model.strip() await db.commit() return RedirectResponse("/devices", status_code=303) @router.post("/devices/{device_kind}/{device_id}/credential") async def assign_credential(device_kind: str, device_id: int, credential_id: int = Form(...), db: AsyncSession = Depends(get_db)): model = OLTDevice if device_kind == "olt" else MikrotikDevice if device_kind == "mikrotik" else None if model is None: raise HTTPException(400, "Jenis perangkat tidak valid") device = await db.get(model, device_id) credential = await db.get(DeviceCredential, credential_id) if not device or not credential: raise HTTPException(404, "Perangkat atau credential tidak ditemukan") device.credential_id = credential.id await db.commit() return RedirectResponse("/devices", status_code=303) @router.post("/devices/credentials") async def create_credential( name: str = Form(...), protocol: str = Form(...), username: str = Form(...), secret: str = Form(...), db: AsyncSession = Depends(get_db), ): from app.core.device_credentials import encrypt_secret db.add(DeviceCredential(name=name.strip(), protocol=protocol.strip(), username=username.strip(), encrypted_secret=encrypt_secret(secret))) await db.commit() return RedirectResponse("/devices", status_code=303) @router.post("/devices/{device_kind}/{device_id}/test-connection") async def test_connection(device_kind: str, device_id: int, db: AsyncSession = Depends(get_db)): model = OLTDevice if device_kind == "olt" else MikrotikDevice if device_kind == "mikrotik" else None if model is None: raise HTTPException(400, "Jenis perangkat tidak valid") device = await db.get(model, device_id) if not device: raise HTTPException(404, "Perangkat tidak ditemukan") port = device.api_port or (22 if device_kind == "olt" else 8728) ok, detail = await _probe_tcp(device.host, port) if ok: return _json_probe(True, f"{device.name}: port management {device.host}:{port} dapat dijangkau pada level TCP. Protokol dan login perangkat belum diuji.") return _json_probe(False, f"{device.name}: port management {device.host}:{port} tidak dapat dijangkau: {detail}") @router.get("/devices/olt/identity", response_class=JSONResponse) async def test_assigned_cdata_ssh_identity(db: AsyncSession = Depends(get_db)): olt = (await db.execute(select(OLTDevice).where(OLTDevice.is_active == True).order_by(OLTDevice.id))).scalars().first() if not olt or not olt.credential_id: return _json_probe(False, "OLT aktif atau credential SSH belum tersedia.") credential = await db.get(DeviceCredential, olt.credential_id) expected = os.getenv(f"OLT_{olt.id}_SSH_HOST_KEY_SHA256") from app.core.ssh_identity import test_pinned_ssh_identity result = test_pinned_ssh_identity(host=olt.host, port=olt.api_port or 22, username=credential.username, encrypted_secret=credential.encrypted_secret, expected_host_key_fingerprint=expected) return JSONResponse({"ok": result.ok, "message": result.message, "fingerprint": result.server_fingerprint}) @router.get("/devices/olt/{olt_id}/onts", response_class=HTMLResponse) async def list_olt_onts(olt_id: int, request: Request, pon_port: str = "0/0/1", db: AsyncSession = Depends(get_db)): olt = await db.get(OLTDevice, olt_id) if not olt: raise HTTPException(404, "OLT tidak ditemukan") cred = await db.get(DeviceCredential, olt.credential_id) if olt.credential_id else None error = None onts = [] if not cred: error = "Credential SSH OLT belum dipasang." else: try: from app.core.cdata_fd1601s_discovery import discover_unregistered_onts onts = discover_unregistered_onts(host=olt.host, port=olt.api_port or 22, username=cred.username, encrypted_secret=cred.encrypted_secret, pinned_host_key="I/zDmJGfTPy3NkMNPghkjawVdeI357pJ/k+xm2qtLdY", pon_port=pon_port) except Exception as exc: error = str(exc) return templates.TemplateResponse(request=request, name="ont_list.html", context={"olt": olt, "onts": onts, "pon_port": pon_port, "error": error})