97 lines
3.3 KiB
Python
97 lines
3.3 KiB
Python
"""Device management — the NAS/AP boxes seen in accounting.
|
|
|
|
Unlike *clients* (customer MACs), a "device" here is a piece of network gear
|
|
identified by its **AP MAC** — the part before ':' in ``radacct.calledstationid``
|
|
(`<AP-MAC>:<SSID>`). Rows are grouped by that MAC, so one physical AP is a single
|
|
row even when it broadcasts several SSIDs (all SSIDs, and the NAS IP(s) it reports
|
|
from, are aggregated). An editable ``alias`` lives in the standalone
|
|
``radadmin_devices`` table, keyed by AP MAC — FreeRADIUS never reads it.
|
|
"""
|
|
from fastapi import APIRouter, Depends
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from ..database import get_db
|
|
from ..models import RadadminDevice, RadAcct
|
|
from ..schemas import DeviceAliasUpdate, DeviceOut
|
|
|
|
router = APIRouter(prefix="/device", tags=["device"])
|
|
|
|
|
|
def _norm_mac(mac: str) -> str:
|
|
"""Canonical AP MAC — uppercase, hyphen-separated (matches client MAC form)."""
|
|
return mac.strip().upper().replace(":", "-")
|
|
|
|
|
|
def _split_called(called: str | None) -> tuple[str | None, str | None]:
|
|
"""radacct.calledstationid is ``<AP-MAC>:<SSID>`` — split into (mac, ssid).
|
|
|
|
The MAC is normalized to uppercase-hyphen; the SSID is left verbatim.
|
|
"""
|
|
if not called:
|
|
return None, None
|
|
if ":" in called:
|
|
mac, ssid = called.split(":", 1)
|
|
return (_norm_mac(mac) or None), (ssid or None)
|
|
return _norm_mac(called) or None, None
|
|
|
|
|
|
def _aliases(db: Session) -> dict[str, str | None]:
|
|
rows = db.execute(select(RadadminDevice.mac_address, RadadminDevice.alias)).all()
|
|
return {mac: alias for mac, alias in rows}
|
|
|
|
|
|
def _devices(db: Session) -> dict[str, dict]:
|
|
"""Aggregate radacct into {ap_mac: {ips, ssids}} over distinct station pairs."""
|
|
pairs = db.execute(
|
|
select(RadAcct.nasipaddress, RadAcct.calledstationid).distinct()
|
|
).all()
|
|
devices: dict[str, dict] = {}
|
|
for ip, called in pairs:
|
|
mac, ssid = _split_called(called)
|
|
if not mac:
|
|
continue
|
|
d = devices.setdefault(mac, {"ips": set(), "ssids": set()})
|
|
if ip:
|
|
d["ips"].add(ip)
|
|
if ssid:
|
|
d["ssids"].add(ssid)
|
|
return devices
|
|
|
|
|
|
def _device_out(ap_mac: str, agg: dict | None, alias: str | None) -> DeviceOut:
|
|
agg = agg or {"ips": set(), "ssids": set()}
|
|
return DeviceOut(
|
|
ap_mac=ap_mac,
|
|
nasipaddress=", ".join(sorted(agg["ips"])) or None,
|
|
ssids=sorted(agg["ssids"]),
|
|
alias=alias,
|
|
)
|
|
|
|
|
|
@router.get("/", response_model=list[DeviceOut])
|
|
def list_devices(db: Session = Depends(get_db)):
|
|
"""List NAS/AP devices — one row per AP MAC, SSIDs and NAS IPs aggregated."""
|
|
devices = _devices(db)
|
|
aliases = _aliases(db)
|
|
return [
|
|
_device_out(mac, devices[mac], aliases.get(mac))
|
|
for mac in sorted(devices)
|
|
]
|
|
|
|
|
|
@router.put("/{ap_mac}", response_model=DeviceOut)
|
|
def set_alias(ap_mac: str, payload: DeviceAliasUpdate, db: Session = Depends(get_db)):
|
|
"""Set (or clear) the alias for an AP MAC — upserts into radadmin_devices."""
|
|
mac = _norm_mac(ap_mac)
|
|
row = db.execute(
|
|
select(RadadminDevice).where(RadadminDevice.mac_address == mac)
|
|
).scalar_one_or_none()
|
|
if row is None:
|
|
db.add(RadadminDevice(mac_address=mac, alias=payload.alias))
|
|
else:
|
|
row.alias = payload.alias
|
|
db.commit()
|
|
|
|
return _device_out(mac, _devices(db).get(mac), payload.alias)
|