113 lines
4.1 KiB
Python
113 lines
4.1 KiB
Python
"""VLAN management — a logical view over the ``radgroupreply`` table.
|
|
|
|
A single "VLAN" is stored as three reply rows sharing one ``groupname`` (the alias):
|
|
|
|
groupname | attribute | op | value
|
|
----------+--------------------------+----+----------
|
|
staff | Tunnel-Type | = | VLAN
|
|
staff | Tunnel-Medium-Type | = | IEEE-802
|
|
staff | Tunnel-Private-Group-Id | = | 55 <- the VLAN ID
|
|
|
|
This router hides that shape behind alias + vlanid.
|
|
"""
|
|
from fastapi import APIRouter, Depends
|
|
from sqlalchemy import delete, select, update
|
|
from sqlalchemy.orm import Session
|
|
|
|
from ..database import get_db
|
|
from ..errors import APIError
|
|
from ..models import RadGroupReply
|
|
from ..schemas import VlanCreate, VlanEdit, VlanOut
|
|
|
|
router = APIRouter(prefix="/vlan", tags=["vlan"])
|
|
|
|
ID_ATTR = "Tunnel-Private-Group-Id"
|
|
OP = "="
|
|
|
|
# (attribute, value) pairs written for every VLAN. `None` value = the VLAN ID.
|
|
VLAN_ROWS: list[tuple[str, str | None]] = [
|
|
("Tunnel-Type", "VLAN"),
|
|
("Tunnel-Medium-Type", "IEEE-802"),
|
|
(ID_ATTR, None),
|
|
]
|
|
|
|
|
|
def _groupnames_for_vlanid(db: Session, vlanid: int) -> list[str]:
|
|
stmt = select(RadGroupReply.groupname).where(
|
|
RadGroupReply.attribute == ID_ATTR,
|
|
RadGroupReply.value == str(vlanid),
|
|
)
|
|
return list(db.execute(stmt).scalars().all())
|
|
|
|
|
|
def _alias_exists(db: Session, alias: str) -> bool:
|
|
stmt = select(RadGroupReply.id).where(RadGroupReply.groupname == alias).limit(1)
|
|
return db.execute(stmt).first() is not None
|
|
|
|
|
|
@router.get("/", response_model=list[VlanOut])
|
|
def list_vlans(db: Session = Depends(get_db)):
|
|
"""List VLANs — one entry per group, showing only alias + VLAN ID."""
|
|
stmt = (
|
|
select(RadGroupReply.groupname, RadGroupReply.value)
|
|
.where(RadGroupReply.attribute == ID_ATTR)
|
|
.order_by(RadGroupReply.value.asc())
|
|
)
|
|
rows = db.execute(stmt).all()
|
|
return [VlanOut(alias=gn, vlanid=int(val)) for gn, val in rows]
|
|
|
|
|
|
@router.post("/add", response_model=VlanOut, status_code=201)
|
|
def add_vlan(payload: VlanCreate, db: Session = Depends(get_db)):
|
|
"""Create a VLAN — inserts the 3 radgroupreply rows atomically."""
|
|
# reject duplicate VLAN ID or duplicate alias
|
|
if _groupnames_for_vlanid(db, payload.vlanid):
|
|
raise APIError(status_code=409, detail=f"VLAN ID {payload.vlanid} already exists")
|
|
if _alias_exists(db, payload.alias):
|
|
raise APIError(status_code=409, detail=f"Alias '{payload.alias}' already exists")
|
|
|
|
for attribute, value in VLAN_ROWS:
|
|
db.add(RadGroupReply(
|
|
groupname=payload.alias,
|
|
attribute=attribute,
|
|
op=OP,
|
|
value=str(payload.vlanid) if value is None else value,
|
|
))
|
|
db.commit()
|
|
return VlanOut(alias=payload.alias, vlanid=payload.vlanid)
|
|
|
|
|
|
@router.post("/edit", response_model=VlanOut)
|
|
def edit_vlan_alias(payload: VlanEdit, db: Session = Depends(get_db)):
|
|
"""Rename a VLAN's alias (groupname), identified by its VLAN ID."""
|
|
groupnames = _groupnames_for_vlanid(db, payload.vlanid)
|
|
if not groupnames:
|
|
raise APIError(status_code=404, detail=f"VLAN ID {payload.vlanid} not found")
|
|
|
|
current = groupnames[0]
|
|
if payload.alias == current:
|
|
return VlanOut(alias=current, vlanid=payload.vlanid)
|
|
|
|
# new alias must not collide with a different group
|
|
if _alias_exists(db, payload.alias):
|
|
raise APIError(status_code=409, detail=f"Alias '{payload.alias}' already exists")
|
|
|
|
db.execute(
|
|
update(RadGroupReply)
|
|
.where(RadGroupReply.groupname == current)
|
|
.values(groupname=payload.alias)
|
|
)
|
|
db.commit()
|
|
return VlanOut(alias=payload.alias, vlanid=payload.vlanid)
|
|
|
|
|
|
@router.delete("/{vlanid}", status_code=204)
|
|
def delete_vlan(vlanid: int, db: Session = Depends(get_db)):
|
|
"""Delete a VLAN — removes every radgroupreply row for the matching group(s)."""
|
|
groupnames = _groupnames_for_vlanid(db, vlanid)
|
|
if not groupnames:
|
|
raise APIError(status_code=404, detail=f"VLAN ID {vlanid} not found")
|
|
|
|
db.execute(delete(RadGroupReply).where(RadGroupReply.groupname.in_(groupnames)))
|
|
db.commit()
|