sarlink/radadmin
6
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

restructure: move backend into backend/

Browse Source
This commit is contained in:
shihaam
2026-07-31 23:53:50 +05:00
parent 9074e55a13
commit 79bc04aaa7
29 changed files with 0 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/__init__.py
View File
backend/app/auth.py
+15
View File
@@ -0,0 +1,15 @@
import secrets
from fastapi import Security
from fastapi.security import APIKeyHeader
from .config import get_settings
from .errors import APIError
api_key_header = APIKeyHeader(name="X-API-Key", auto_error=False)
def require_api_key(api_key: str | None = Security(api_key_header)) -> None:
expected = get_settings().api_key
if not api_key or not secrets.compare_digest(api_key, expected):
raise APIError(status_code=401, detail="Invalid or missing API key")
backend/app/config.py
+36
View File
@@ -0,0 +1,36 @@
from functools import lru_cache
from pydantic_settings import BaseSettings, SettingsConfigDict
class Settings(BaseSettings):
model_config = SettingsConfigDict(env_file=".env", env_file_encoding="utf-8", extra="ignore")
db_host: str = "127.0.0.1"
db_port: int = 3306
db_user: str = "root"
db_password: str = ""
db_name: str = "radius"
api_key: str = "change-me"
cors_origins: str = "*"
default_limit: int = 50
max_limit: int = 500
@property
def database_url(self) -> str:
from urllib.parse import quote_plus
return (
f"mysql+pymysql://{self.db_user}:{quote_plus(self.db_password)}"
f"@{self.db_host}:{self.db_port}/{self.db_name}?charset=utf8mb4"
)
@property
def cors_origin_list(self) -> list[str]:
return [o.strip() for o in self.cors_origins.split(",") if o.strip()]
@lru_cache
def get_settings() -> Settings:
return Settings()
backend/app/crud.py
+79
View File
@@ -0,0 +1,79 @@
from typing import Any
from sqlalchemy import func, select
from sqlalchemy.orm import Session
from .database import Base
from .errors import APIError
def list_rows(
db: Session,
model: type[Base],
*,
limit: int,
offset: int,
filters: dict[str, Any] | None = None,
order_by: Any | None = None,
) -> list[Base]:
stmt = select(model)
for field, value in (filters or {}).items():
if value is not None:
stmt = stmt.where(getattr(model, field) == value)
if order_by is not None:
stmt = stmt.order_by(order_by)
stmt = stmt.limit(limit).offset(offset)
return list(db.execute(stmt).scalars().all())
def count_rows(db: Session, model: type[Base], filters: dict[str, Any] | None = None) -> int:
stmt = select(func.count()).select_from(model)
for field, value in (filters or {}).items():
if value is not None:
stmt = stmt.where(getattr(model, field) == value)
return int(db.execute(stmt).scalar_one())
def get_row_or_404(db: Session, model: type[Base], pk: Any, pk_field: str = "id") -> Base:
obj = db.execute(select(model).where(getattr(model, pk_field) == pk)).scalar_one_or_none()
if obj is None:
raise APIError(status_code=404, detail=f"{model.__tablename__} '{pk}' not found")
return obj
def create_row(db: Session, model: type[Base], data: dict[str, Any]) -> Base:
obj = model(**data)
db.add(obj)
_commit(db)
db.refresh(obj)
return obj
def update_row(db: Session, obj: Base, data: dict[str, Any]) -> Base:
for field, value in data.items():
setattr(obj, field, value)
_commit(db)
db.refresh(obj)
return obj
def delete_row(db: Session, obj: Base) -> None:
db.delete(obj)
_commit(db)
def _commit(db: Session) -> None:
from sqlalchemy.exc import IntegrityError
try:
db.commit()
except IntegrityError as exc:
db.rollback()
raise APIError(status_code=409, detail=_integrity_message(exc)) from exc
def _integrity_message(exc: Exception) -> str:
msg = str(getattr(exc, "orig", exc))
if "Duplicate entry" in msg:
return "Duplicate entry: a record with these unique values already exists"
return "Database integrity error"
backend/app/database.py
+29
View File
@@ -0,0 +1,29 @@
from collections.abc import Generator
from sqlalchemy import create_engine
from sqlalchemy.orm import DeclarativeBase, Session, sessionmaker
from .config import get_settings
settings = get_settings()
engine = create_engine(
settings.database_url,
pool_pre_ping=True,
pool_recycle=3600,
future=True,
)
SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False, future=True)
class Base(DeclarativeBase):
pass
def get_db() -> Generator[Session, None, None]:
db = SessionLocal()
try:
yield db
finally:
db.close()
backend/app/errors.py
+15
View File
@@ -0,0 +1,15 @@
from fastapi import Request
from fastapi.responses import JSONResponse
class APIError(Exception):
"""Application error that maps to a JSON response with a status code."""
def __init__(self, status_code: int, detail: str):
self.status_code = status_code
self.detail = detail
super().__init__(detail)
async def api_error_handler(request: Request, exc: APIError) -> JSONResponse:
return JSONResponse(status_code=exc.status_code, content={"detail": exc.detail})
backend/app/main.py
+77
View File
@@ -0,0 +1,77 @@
from fastapi import Depends, FastAPI
from fastapi.middleware.cors import CORSMiddleware
from sqlalchemy import text
from .auth import require_api_key
from .config import get_settings
from .database import engine
from .errors import APIError, api_error_handler
from .routers import (
customers,
device,
nas,
nasreload,
radacct,
radcheck,
radgroupcheck,
radgroupreply,
radpostauth,
radreply,
radusergroup,
vlan,
)
settings = get_settings()
app = FastAPI(
title="FreeRADIUS REST API",
description="RESTful CRUD over the FreeRADIUS SQL schema. "
"All endpoints require the `X-API-Key` header.",
version="1.0.0",
)
app.add_middleware(
CORSMiddleware,
allow_origins=settings.cors_origin_list,
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
app.add_exception_handler(APIError, api_error_handler)
# ---- unauthenticated meta endpoints ----
@app.get("/", tags=["meta"])
def root():
return {"service": "freeradius-rest-api", "docs": "/docs", "health": "/health"}
@app.get("/health", tags=["meta"])
def health():
try:
with engine.connect() as conn:
conn.execute(text("SELECT 1"))
db_ok = True
except Exception:
db_ok = False
return {"status": "ok" if db_ok else "degraded", "database": "up" if db_ok else "down"}
# ---- authenticated resource routers ----
protected = [Depends(require_api_key)]
for module in (
customers,
nas,
radcheck,
radreply,
radgroupcheck,
radgroupreply,
radusergroup,
vlan,
device,
radacct,
radpostauth,
nasreload,
):
app.include_router(module.router, dependencies=protected)
backend/app/models.py
+136
View File
@@ -0,0 +1,136 @@
from datetime import datetime
from sqlalchemy import BigInteger, DateTime, Integer, String, func
from sqlalchemy.orm import Mapped, mapped_column
from .database import Base
class Customer(Base):
__tablename__ = "customers"
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
username: Mapped[str] = mapped_column(String(64), nullable=False)
mac_address: Mapped[str] = mapped_column(String(17), nullable=False)
status: Mapped[str] = mapped_column(String(10), nullable=False, default="new")
name: Mapped[str | None] = mapped_column(String(128), nullable=True)
phone: Mapped[str | None] = mapped_column(String(32), nullable=True)
device_alias: Mapped[str | None] = mapped_column(String(64), nullable=True)
created_at: Mapped[datetime | None] = mapped_column(
DateTime, nullable=True, server_default=func.current_timestamp()
)
class Nas(Base):
__tablename__ = "nas"
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
nasname: Mapped[str] = mapped_column(String(128), nullable=False)
shortname: Mapped[str | None] = mapped_column(String(32), nullable=True)
type: Mapped[str | None] = mapped_column(String(30), nullable=True, default="other")
ports: Mapped[int | None] = mapped_column(Integer, nullable=True)
secret: Mapped[str] = mapped_column(String(60), nullable=False, default="secret")
server: Mapped[str | None] = mapped_column(String(64), nullable=True)
community: Mapped[str | None] = mapped_column(String(50), nullable=True)
description: Mapped[str | None] = mapped_column(String(200), nullable=True, default="RADIUS Client")
class NasReload(Base):
__tablename__ = "nasreload"
nasipaddress: Mapped[str] = mapped_column(String(15), primary_key=True)
reloadtime: Mapped[datetime] = mapped_column(DateTime, nullable=False)
class RadCheck(Base):
__tablename__ = "radcheck"
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
username: Mapped[str] = mapped_column(String(64), nullable=False, default="")
attribute: Mapped[str] = mapped_column(String(64), nullable=False, default="")
op: Mapped[str] = mapped_column(String(2), nullable=False, default="==")
value: Mapped[str] = mapped_column(String(253), nullable=False, default="")
class RadReply(Base):
__tablename__ = "radreply"
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
username: Mapped[str] = mapped_column(String(64), nullable=False, default="")
attribute: Mapped[str] = mapped_column(String(64), nullable=False, default="")
op: Mapped[str] = mapped_column(String(2), nullable=False, default="=")
value: Mapped[str] = mapped_column(String(253), nullable=False, default="")
class RadGroupCheck(Base):
__tablename__ = "radgroupcheck"
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
groupname: Mapped[str] = mapped_column(String(64), nullable=False, default="")
attribute: Mapped[str] = mapped_column(String(64), nullable=False, default="")
op: Mapped[str] = mapped_column(String(2), nullable=False, default="==")
value: Mapped[str] = mapped_column(String(253), nullable=False, default="")
class RadGroupReply(Base):
__tablename__ = "radgroupreply"
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
groupname: Mapped[str] = mapped_column(String(64), nullable=False, default="")
attribute: Mapped[str] = mapped_column(String(64), nullable=False, default="")
op: Mapped[str] = mapped_column(String(2), nullable=False, default="=")
value: Mapped[str] = mapped_column(String(253), nullable=False, default="")
class RadUserGroup(Base):
__tablename__ = "radusergroup"
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
username: Mapped[str] = mapped_column(String(64), nullable=False, default="")
groupname: Mapped[str] = mapped_column(String(64), nullable=False, default="")
priority: Mapped[int] = mapped_column(Integer, nullable=False, default=1)
class RadAcct(Base):
__tablename__ = "radacct"
radacctid: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)
acctsessionid: Mapped[str] = mapped_column(String(64), nullable=False, default="")
acctuniqueid: Mapped[str] = mapped_column(String(32), nullable=False, default="")
username: Mapped[str] = mapped_column(String(64), nullable=False, default="")
realm: Mapped[str | None] = mapped_column(String(64), nullable=True, default="")
nasipaddress: Mapped[str] = mapped_column(String(15), nullable=False, default="")
nasportid: Mapped[str | None] = mapped_column(String(32), nullable=True)
nasporttype: Mapped[str | None] = mapped_column(String(32), nullable=True)
acctstarttime: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)
acctupdatetime: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)
acctstoptime: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)
acctinterval: Mapped[int | None] = mapped_column(Integer, nullable=True)
acctsessiontime: Mapped[int | None] = mapped_column(Integer, nullable=True)
acctauthentic: Mapped[str | None] = mapped_column(String(32), nullable=True)
connectinfo_start: Mapped[str | None] = mapped_column(String(128), nullable=True)
connectinfo_stop: Mapped[str | None] = mapped_column(String(128), nullable=True)
acctinputoctets: Mapped[int | None] = mapped_column(BigInteger, nullable=True)
acctoutputoctets: Mapped[int | None] = mapped_column(BigInteger, nullable=True)
calledstationid: Mapped[str] = mapped_column(String(50), nullable=False, default="")
callingstationid: Mapped[str] = mapped_column(String(50), nullable=False, default="")
acctterminatecause: Mapped[str] = mapped_column(String(32), nullable=False, default="")
servicetype: Mapped[str | None] = mapped_column(String(32), nullable=True)
framedprotocol: Mapped[str | None] = mapped_column(String(32), nullable=True)
framedipaddress: Mapped[str] = mapped_column(String(15), nullable=False, default="")
framedipv6address: Mapped[str] = mapped_column(String(45), nullable=False, default="")
framedipv6prefix: Mapped[str] = mapped_column(String(45), nullable=False, default="")
framedinterfaceid: Mapped[str] = mapped_column(String(44), nullable=False, default="")
delegatedipv6prefix: Mapped[str] = mapped_column(String(45), nullable=False, default="")
class_: Mapped[str | None] = mapped_column("class", String(64), nullable=True)
class RadPostAuth(Base):
__tablename__ = "radpostauth"
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
username: Mapped[str] = mapped_column(String(64), nullable=False, default="")
pass_: Mapped[str] = mapped_column("pass", String(64), nullable=False, default="")
reply: Mapped[str] = mapped_column(String(32), nullable=False, default="")
authdate: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)
class_: Mapped[str | None] = mapped_column("class", String(64), nullable=True)
backend/app/pagination.py
+30
View File
@@ -0,0 +1,30 @@
from typing import Generic, TypeVar
from fastapi import Query
from pydantic import BaseModel
from .config import get_settings
T = TypeVar("T")
class Page(BaseModel, Generic[T]):
total: int
limit: int
offset: int
items: list[T]
class PageParams:
"""Reusable dependency for `?limit=&offset=` with env-configured caps."""
def __init__(
self,
limit: int = Query(default=None, ge=1, description="Max rows to return"),
offset: int = Query(default=0, ge=0, description="Rows to skip"),
):
settings = get_settings()
if limit is None:
limit = settings.default_limit
self.limit = min(limit, settings.max_limit)
self.offset = offset
backend/app/routers/__init__.py
View File
backend/app/routers/_attr_factory.py
+72
View File
@@ -0,0 +1,72 @@
"""Router factories for the AVP-style tables (radcheck/radreply/radgroup*).
They share the same shape — an id plus (username|groupname, attribute, op, value).
Two factories cover the two owner shapes: user-owned (radcheck, radreply) and
group-owned (radgroupcheck, radgroupreply).
"""
from fastapi import APIRouter, Depends
from sqlalchemy.orm import Session
from .. import crud
from ..database import Base, get_db
from ..pagination import Page, PageParams
def build_user_attr_router(*, model, prefix, tag, out_schema, create_schema, update_schema) -> APIRouter:
router = APIRouter(prefix=prefix, tags=[tag])
@router.get("", response_model=Page[out_schema])
def list_items(
page: PageParams = Depends(),
username: str | None = None,
attribute: str | None = None,
db: Session = Depends(get_db),
):
filters = {"username": username, "attribute": attribute}
total = crud.count_rows(db, model, filters)
rows = crud.list_rows(db, model, limit=page.limit, offset=page.offset,
filters=filters, order_by=model.id.asc())
return Page(total=total, limit=page.limit, offset=page.offset, items=rows)
_register_item_routes(router, model, out_schema, create_schema, update_schema)
return router
def build_group_attr_router(*, model, prefix, tag, out_schema, create_schema, update_schema) -> APIRouter:
router = APIRouter(prefix=prefix, tags=[tag])
@router.get("", response_model=Page[out_schema])
def list_items(
page: PageParams = Depends(),
groupname: str | None = None,
attribute: str | None = None,
db: Session = Depends(get_db),
):
filters = {"groupname": groupname, "attribute": attribute}
total = crud.count_rows(db, model, filters)
rows = crud.list_rows(db, model, limit=page.limit, offset=page.offset,
filters=filters, order_by=model.id.asc())
return Page(total=total, limit=page.limit, offset=page.offset, items=rows)
_register_item_routes(router, model, out_schema, create_schema, update_schema)
return router
def _register_item_routes(router: APIRouter, model: type[Base], out_schema, create_schema, update_schema) -> None:
@router.get("/{item_id}", response_model=out_schema)
def get_item(item_id: int, db: Session = Depends(get_db)):
return crud.get_row_or_404(db, model, item_id)
@router.post("", response_model=out_schema, status_code=201)
def create_item(payload: create_schema, db: Session = Depends(get_db)):
return crud.create_row(db, model, payload.model_dump())
@router.put("/{item_id}", response_model=out_schema)
def update_item(item_id: int, payload: update_schema, db: Session = Depends(get_db)):
obj = crud.get_row_or_404(db, model, item_id)
return crud.update_row(db, obj, payload.model_dump(exclude_unset=True))
@router.delete("/{item_id}", status_code=204)
def delete_item(item_id: int, db: Session = Depends(get_db)):
obj = crud.get_row_or_404(db, model, item_id)
crud.delete_row(db, obj)
backend/app/routers/customers.py
+47
View File
@@ -0,0 +1,47 @@
from fastapi import APIRouter, Depends
from sqlalchemy.orm import Session
from .. import crud
from ..database import get_db
from ..models import Customer
from ..pagination import Page, PageParams
from ..schemas import CustomerCreate, CustomerOut, CustomerUpdate
router = APIRouter(prefix="/customers", tags=["customers"])
@router.get("", response_model=Page[CustomerOut])
def list_customers(
page: PageParams = Depends(),
username: str | None = None,
mac_address: str | None = None,
status: str | None = None,
db: Session = Depends(get_db),
):
filters = {"username": username, "mac_address": mac_address, "status": status}
total = crud.count_rows(db, Customer, filters)
rows = crud.list_rows(db, Customer, limit=page.limit, offset=page.offset, filters=filters,
order_by=Customer.id.desc())
return Page(total=total, limit=page.limit, offset=page.offset, items=rows)
@router.get("/{customer_id}", response_model=CustomerOut)
def get_customer(customer_id: int, db: Session = Depends(get_db)):
return crud.get_row_or_404(db, Customer, customer_id)
@router.post("", response_model=CustomerOut, status_code=201)
def create_customer(payload: CustomerCreate, db: Session = Depends(get_db)):
return crud.create_row(db, Customer, payload.model_dump())
@router.put("/{customer_id}", response_model=CustomerOut)
def update_customer(customer_id: int, payload: CustomerUpdate, db: Session = Depends(get_db)):
obj = crud.get_row_or_404(db, Customer, customer_id)
return crud.update_row(db, obj, payload.model_dump(exclude_unset=True))
@router.delete("/{customer_id}", status_code=204)
def delete_customer(customer_id: int, db: Session = Depends(get_db)):
obj = crud.get_row_or_404(db, Customer, customer_id)
crud.delete_row(db, obj)
backend/app/routers/device.py
+139
View File
@@ -0,0 +1,139 @@
"""Device management — a logical view spanning three tables.
A "device" is a client identified by its MAC address, which is used verbatim as
the RADIUS ``username``. One device touches three tables:
radcheck username = MAC, Cleartext-Password := MAC (auth)
radusergroup username = MAC, groupname = <group> (VLAN/group membership)
customers username = MAC, mac_address = MAC, status (billing/metadata)
This router hides that fan-out behind mac_address + group + status.
"""
from fastapi import APIRouter, Depends
from sqlalchemy import delete, func, select, update
from sqlalchemy.orm import Session
from ..database import get_db
from ..errors import APIError
from ..models import Customer, RadCheck, RadGroupReply, RadUserGroup
from ..pagination import Page, PageParams
from ..schemas import DeviceCreate, DeviceEdit, DeviceOut
router = APIRouter(prefix="/device", tags=["device"])
def _group_exists(db: Session, group: str) -> bool:
stmt = select(RadGroupReply.id).where(RadGroupReply.groupname == group).limit(1)
return db.execute(stmt).first() is not None
def _device_exists(db: Session, mac: str) -> bool:
stmt = select(Customer.id).where(Customer.username == mac).limit(1)
return db.execute(stmt).first() is not None
@router.get("/", response_model=Page[DeviceOut])
def list_devices(page: PageParams = Depends(), db: Session = Depends(get_db)):
"""List devices — MAC, group and status, joined from customers + radusergroup."""
base = select(
Customer.mac_address, RadUserGroup.groupname, Customer.status,
Customer.name, Customer.phone, Customer.device_alias,
).outerjoin(RadUserGroup, RadUserGroup.username == Customer.username)
total = db.execute(select(func.count()).select_from(Customer)).scalar_one()
rows = db.execute(base.order_by(Customer.id.desc()).limit(page.limit).offset(page.offset)).all()
items = [
DeviceOut(mac_address=mac, group=gn, status=st, name=nm, phone=ph, alias=al)
for mac, gn, st, nm, ph, al in rows
]
return Page(total=int(total), limit=page.limit, offset=page.offset, items=items)
@router.get("/{mac_address}", response_model=DeviceOut)
def get_device(mac_address: str, db: Session = Depends(get_db)):
mac = mac_address.strip().upper().replace(":", "-")
stmt = (
select(
Customer.mac_address, RadUserGroup.groupname, Customer.status,
Customer.name, Customer.phone, Customer.device_alias,
)
.outerjoin(RadUserGroup, RadUserGroup.username == Customer.username)
.where(Customer.username == mac)
)
row = db.execute(stmt).first()
if row is None:
raise APIError(status_code=404, detail=f"Device '{mac}' not found")
return DeviceOut(
mac_address=row[0], group=row[1], status=row[2],
name=row[3], phone=row[4], alias=row[5],
)
@router.post("/add", response_model=DeviceOut, status_code=201)
def add_device(payload: DeviceCreate, db: Session = Depends(get_db)):
"""Register a device: create its radcheck, radusergroup and customer rows."""
mac = payload.mac_address
if not _group_exists(db, payload.group):
raise APIError(status_code=400, detail=f"Group '{payload.group}' not found — create it first")
if _device_exists(db, mac):
raise APIError(status_code=409, detail=f"Device '{mac}' already exists")
db.add(RadCheck(username=mac, attribute="Cleartext-Password", op=":=", value=mac))
db.add(RadUserGroup(username=mac, groupname=payload.group, priority=1))
db.add(Customer(
username=mac, mac_address=mac, status="paid",
name=payload.name, phone=payload.phone, device_alias=payload.alias,
))
db.commit()
return DeviceOut(
mac_address=mac, group=payload.group, status="paid",
name=payload.name, phone=payload.phone, alias=payload.alias,
)
@router.post("/edit", response_model=DeviceOut)
def edit_device(payload: DeviceEdit, db: Session = Depends(get_db)):
"""Edit a device — any subset of group, status, name, phone, alias."""
mac = payload.mac_address
customer = db.execute(select(Customer).where(Customer.username == mac)).scalar_one_or_none()
if customer is None:
raise APIError(status_code=404, detail=f"Device '{mac}' not found")
if payload.group is not None:
if not _group_exists(db, payload.group):
raise APIError(status_code=400, detail=f"Group '{payload.group}' not found")
db.execute(
update(RadUserGroup).where(RadUserGroup.username == mac).values(groupname=payload.group)
)
if payload.status is not None:
customer.status = payload.status
if payload.name is not None:
customer.name = payload.name
if payload.phone is not None:
customer.phone = payload.phone
if payload.alias is not None:
customer.device_alias = payload.alias
db.commit()
group = db.execute(
select(RadUserGroup.groupname).where(RadUserGroup.username == mac).limit(1)
).scalar_one_or_none()
return DeviceOut(
mac_address=mac, group=group, status=customer.status,
name=customer.name, phone=customer.phone, alias=customer.device_alias,
)
@router.delete("/{mac_address}", status_code=204)
def delete_device(mac_address: str, db: Session = Depends(get_db)):
"""Remove a device from radcheck, radusergroup and customers."""
mac = mac_address.strip().upper().replace(":", "-")
if not _device_exists(db, mac):
raise APIError(status_code=404, detail=f"Device '{mac}' not found")
db.execute(delete(RadCheck).where(RadCheck.username == mac))
db.execute(delete(RadUserGroup).where(RadUserGroup.username == mac))
db.execute(delete(Customer).where(Customer.username == mac))
db.commit()
backend/app/routers/nas.py
+46
View File
@@ -0,0 +1,46 @@
from fastapi import APIRouter, Depends
from sqlalchemy.orm import Session
from .. import crud
from ..database import get_db
from ..models import Nas
from ..pagination import Page, PageParams
from ..schemas import NasCreate, NasOut, NasUpdate
router = APIRouter(prefix="/nas", tags=["nas"])
@router.get("", response_model=Page[NasOut])
def list_nas(
page: PageParams = Depends(),
nasname: str | None = None,
shortname: str | None = None,
db: Session = Depends(get_db),
):
filters = {"nasname": nasname, "shortname": shortname}
total = crud.count_rows(db, Nas, filters)
rows = crud.list_rows(db, Nas, limit=page.limit, offset=page.offset, filters=filters,
order_by=Nas.id.asc())
return Page(total=total, limit=page.limit, offset=page.offset, items=rows)
@router.get("/{nas_id}", response_model=NasOut)
def get_nas(nas_id: int, db: Session = Depends(get_db)):
return crud.get_row_or_404(db, Nas, nas_id)
@router.post("", response_model=NasOut, status_code=201)
def create_nas(payload: NasCreate, db: Session = Depends(get_db)):
return crud.create_row(db, Nas, payload.model_dump())
@router.put("/{nas_id}", response_model=NasOut)
def update_nas(nas_id: int, payload: NasUpdate, db: Session = Depends(get_db)):
obj = crud.get_row_or_404(db, Nas, nas_id)
return crud.update_row(db, obj, payload.model_dump(exclude_unset=True))
@router.delete("/{nas_id}", status_code=204)
def delete_nas(nas_id: int, db: Session = Depends(get_db)):
obj = crud.get_row_or_404(db, Nas, nas_id)
crud.delete_row(db, obj)
backend/app/routers/nasreload.py
+24
View File
@@ -0,0 +1,24 @@
"""nasreload — last reload time per NAS. Read-only."""
from fastapi import APIRouter, Depends
from sqlalchemy.orm import Session
from .. import crud
from ..database import get_db
from ..models import NasReload
from ..pagination import Page, PageParams
from ..schemas import NasReloadOut
router = APIRouter(prefix="/nasreload", tags=["nasreload (read-only)"])
@router.get("", response_model=Page[NasReloadOut])
def list_nasreload(page: PageParams = Depends(), db: Session = Depends(get_db)):
total = crud.count_rows(db, NasReload)
rows = crud.list_rows(db, NasReload, limit=page.limit, offset=page.offset,
order_by=NasReload.reloadtime.desc())
return Page(total=total, limit=page.limit, offset=page.offset, items=rows)
@router.get("/{nasipaddress}", response_model=NasReloadOut)
def get_nasreload(nasipaddress: str, db: Session = Depends(get_db)):
return crud.get_row_or_404(db, NasReload, nasipaddress, pk_field="nasipaddress")
backend/app/routers/radacct.py
+41
View File
@@ -0,0 +1,41 @@
"""radacct — accounting data. Read-only: FreeRADIUS owns writes to this table."""
from fastapi import APIRouter, Depends, Query
from sqlalchemy import select
from sqlalchemy.orm import Session
from .. import crud
from ..database import get_db
from ..models import RadAcct
from ..pagination import Page, PageParams
from ..schemas import RadAcctOut
router = APIRouter(prefix="/radacct", tags=["radacct (read-only)"])
@router.get("", response_model=Page[RadAcctOut])
def list_acct(
page: PageParams = Depends(),
username: str | None = None,
nasipaddress: str | None = None,
active: bool | None = Query(default=None, description="true = sessions with no stop time"),
db: Session = Depends(get_db),
):
filters = {"username": username, "nasipaddress": nasipaddress}
total = crud.count_rows(db, RadAcct, filters)
stmt = select(RadAcct)
for field, value in filters.items():
if value is not None:
stmt = stmt.where(getattr(RadAcct, field) == value)
if active is True:
stmt = stmt.where(RadAcct.acctstoptime.is_(None))
elif active is False:
stmt = stmt.where(RadAcct.acctstoptime.is_not(None))
stmt = stmt.order_by(RadAcct.acctstarttime.desc()).limit(page.limit).offset(page.offset)
rows = list(db.execute(stmt).scalars().all())
return Page(total=total, limit=page.limit, offset=page.offset, items=rows)
@router.get("/{radacctid}", response_model=RadAcctOut)
def get_acct(radacctid: int, db: Session = Depends(get_db)):
return crud.get_row_or_404(db, RadAcct, radacctid, pk_field="radacctid")
backend/app/routers/radcheck.py
+12
View File
@@ -0,0 +1,12 @@
from ..models import RadCheck
from ..schemas import UserAttrCreate, UserAttrOut, UserAttrUpdate
from ._attr_factory import build_user_attr_router
router = build_user_attr_router(
model=RadCheck,
prefix="/radcheck",
tag="radcheck",
out_schema=UserAttrOut,
create_schema=UserAttrCreate,
update_schema=UserAttrUpdate,
)
backend/app/routers/radgroupcheck.py
+12
View File
@@ -0,0 +1,12 @@
from ..models import RadGroupCheck
from ..schemas import GroupAttrCreate, GroupAttrOut, GroupAttrUpdate
from ._attr_factory import build_group_attr_router
router = build_group_attr_router(
model=RadGroupCheck,
prefix="/radgroupcheck",
tag="radgroupcheck",
out_schema=GroupAttrOut,
create_schema=GroupAttrCreate,
update_schema=GroupAttrUpdate,
)
backend/app/routers/radgroupreply.py
+17
View File
@@ -0,0 +1,17 @@
"""radgroupreply — raw group reply attributes (all attributes, not just VLAN).
The VLAN-specific abstraction lives in ``routers/vlan.py`` (``/vlan``); this router
stays as generic CRUD over the raw table for any other reply attributes.
"""
from ..models import RadGroupReply
from ..schemas import GroupAttrCreate, GroupAttrOut, GroupAttrUpdate
from ._attr_factory import build_group_attr_router
router = build_group_attr_router(
model=RadGroupReply,
prefix="/radgroupreply",
tag="radgroupreply",
out_schema=GroupAttrOut,
create_schema=GroupAttrCreate,
update_schema=GroupAttrUpdate,
)
backend/app/routers/radpostauth.py
+30
View File
@@ -0,0 +1,30 @@
"""radpostauth — authentication log. Read-only."""
from fastapi import APIRouter, Depends
from sqlalchemy.orm import Session
from .. import crud
from ..database import get_db
from ..models import RadPostAuth
from ..pagination import Page, PageParams
from ..schemas import RadPostAuthOut
router = APIRouter(prefix="/radpostauth", tags=["radpostauth (read-only)"])
@router.get("", response_model=Page[RadPostAuthOut])
def list_postauth(
page: PageParams = Depends(),
username: str | None = None,
reply: str | None = None,
db: Session = Depends(get_db),
):
filters = {"username": username, "reply": reply}
total = crud.count_rows(db, RadPostAuth, filters)
rows = crud.list_rows(db, RadPostAuth, limit=page.limit, offset=page.offset,
filters=filters, order_by=RadPostAuth.id.desc())
return Page(total=total, limit=page.limit, offset=page.offset, items=rows)
@router.get("/{item_id}", response_model=RadPostAuthOut)
def get_postauth(item_id: int, db: Session = Depends(get_db)):
return crud.get_row_or_404(db, RadPostAuth, item_id)
backend/app/routers/radreply.py
+12
View File
@@ -0,0 +1,12 @@
from ..models import RadReply
from ..schemas import UserAttrCreate, UserAttrOut, UserAttrUpdate
from ._attr_factory import build_user_attr_router
router = build_user_attr_router(
model=RadReply,
prefix="/radreply",
tag="radreply",
out_schema=UserAttrOut,
create_schema=UserAttrCreate,
update_schema=UserAttrUpdate,
)
backend/app/routers/radusergroup.py
+46
View File
@@ -0,0 +1,46 @@
from fastapi import APIRouter, Depends
from sqlalchemy.orm import Session
from .. import crud
from ..database import get_db
from ..models import RadUserGroup
from ..pagination import Page, PageParams
from ..schemas import UserGroupCreate, UserGroupOut, UserGroupUpdate
router = APIRouter(prefix="/radusergroup", tags=["radusergroup"])
@router.get("", response_model=Page[UserGroupOut])
def list_usergroups(
page: PageParams = Depends(),
username: str | None = None,
groupname: str | None = None,
db: Session = Depends(get_db),
):
filters = {"username": username, "groupname": groupname}
total = crud.count_rows(db, RadUserGroup, filters)
rows = crud.list_rows(db, RadUserGroup, limit=page.limit, offset=page.offset,
filters=filters, order_by=RadUserGroup.priority.asc())
return Page(total=total, limit=page.limit, offset=page.offset, items=rows)
@router.get("/{item_id}", response_model=UserGroupOut)
def get_usergroup(item_id: int, db: Session = Depends(get_db)):
return crud.get_row_or_404(db, RadUserGroup, item_id)
@router.post("", response_model=UserGroupOut, status_code=201)
def create_usergroup(payload: UserGroupCreate, db: Session = Depends(get_db)):
return crud.create_row(db, RadUserGroup, payload.model_dump())
@router.put("/{item_id}", response_model=UserGroupOut)
def update_usergroup(item_id: int, payload: UserGroupUpdate, db: Session = Depends(get_db)):
obj = crud.get_row_or_404(db, RadUserGroup, item_id)
return crud.update_row(db, obj, payload.model_dump(exclude_unset=True))
@router.delete("/{item_id}", status_code=204)
def delete_usergroup(item_id: int, db: Session = Depends(get_db)):
obj = crud.get_row_or_404(db, RadUserGroup, item_id)
crud.delete_row(db, obj)
backend/app/routers/vlan.py
+112
View File
@@ -0,0 +1,112 @@
"""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()
backend/app/schemas.py
+274
View File
@@ -0,0 +1,274 @@
import re
from datetime import datetime
from typing import Literal
from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
class ORMModel(BaseModel):
model_config = ConfigDict(from_attributes=True)
# ---------- customers ----------
class CustomerBase(BaseModel):
username: str = Field(max_length=64)
mac_address: str = Field(max_length=17)
status: Literal["new", "paid", "unpaid"] = "new"
name: str | None = Field(default=None, max_length=128)
phone: str | None = Field(default=None, max_length=32)
device_alias: str | None = Field(default=None, max_length=64)
class CustomerCreate(CustomerBase):
pass
class CustomerUpdate(BaseModel):
username: str | None = Field(default=None, max_length=64)
mac_address: str | None = Field(default=None, max_length=17)
status: Literal["new", "paid", "unpaid"] | None = None
name: str | None = Field(default=None, max_length=128)
phone: str | None = Field(default=None, max_length=32)
device_alias: str | None = Field(default=None, max_length=64)
class CustomerOut(ORMModel, CustomerBase):
id: int
created_at: datetime | None = None
# ---------- nas ----------
class NasBase(BaseModel):
nasname: str = Field(max_length=128)
shortname: str | None = Field(default=None, max_length=32)
type: str | None = Field(default="other", max_length=30)
ports: int | None = None
secret: str = Field(default="secret", max_length=60)
server: str | None = Field(default=None, max_length=64)
community: str | None = Field(default=None, max_length=50)
description: str | None = Field(default="RADIUS Client", max_length=200)
class NasCreate(NasBase):
pass
class NasUpdate(BaseModel):
nasname: str | None = Field(default=None, max_length=128)
shortname: str | None = Field(default=None, max_length=32)
type: str | None = Field(default=None, max_length=30)
ports: int | None = None
secret: str | None = Field(default=None, max_length=60)
server: str | None = Field(default=None, max_length=64)
community: str | None = Field(default=None, max_length=50)
description: str | None = Field(default=None, max_length=200)
class NasOut(ORMModel, NasBase):
id: int
# ---------- attribute pair tables (radcheck / radreply) ----------
class UserAttrBase(BaseModel):
username: str = Field(max_length=64)
attribute: str = Field(max_length=64)
op: str = Field(max_length=2)
value: str = Field(max_length=253)
class UserAttrCreate(UserAttrBase):
pass
class UserAttrUpdate(BaseModel):
username: str | None = Field(default=None, max_length=64)
attribute: str | None = Field(default=None, max_length=64)
op: str | None = Field(default=None, max_length=2)
value: str | None = Field(default=None, max_length=253)
class UserAttrOut(ORMModel, UserAttrBase):
id: int
# ---------- group attribute tables (radgroupcheck / radgroupreply / vlans) ----------
class GroupAttrBase(BaseModel):
groupname: str = Field(max_length=64)
attribute: str = Field(max_length=64)
op: str = Field(max_length=2)
value: str = Field(max_length=253)
class GroupAttrCreate(GroupAttrBase):
pass
class GroupAttrUpdate(BaseModel):
groupname: str | None = Field(default=None, max_length=64)
attribute: str | None = Field(default=None, max_length=64)
op: str | None = Field(default=None, max_length=2)
value: str | None = Field(default=None, max_length=253)
class GroupAttrOut(ORMModel, GroupAttrBase):
id: int
# ---------- vlans (logical view over radgroupreply) ----------
class VlanOut(BaseModel):
alias: str # radgroupreply.groupname
vlanid: int # value of the Tunnel-Private-Group-Id attribute
class VlanCreate(BaseModel):
vlanid: int = Field(ge=1, le=4094, description="802.1Q VLAN ID")
alias: str = Field(min_length=1, max_length=64, description="Group name for this VLAN")
class VlanEdit(BaseModel):
vlanid: int = Field(ge=1, le=4094, description="VLAN ID identifying the group to rename")
alias: str = Field(min_length=1, max_length=64, description="New alias (groupname)")
# ---------- devices (logical view over radcheck + radusergroup + customers) ----------
_MAC_RE = r"^[0-9A-Fa-f]{2}([-:][0-9A-Fa-f]{2}){5}$"
def _normalize_mac(mac: str) -> str:
"""Canonicalize a MAC to uppercase, hyphen-separated (AA-BB-CC-DD-EE-FF)."""
return mac.strip().upper().replace(":", "-")
_PHONE_ERROR = "Enter a valid phone number"
def _normalize_phone(raw: str) -> str:
"""Validate a Maldivian mobile number and return the 7-digit local form.
Rules: 7 digits, starts with 9 or 7. A 960/+960 country code is stripped ONLY
when the number is 10 digits (960 + 7) — a bare 7-digit number like 9601234 is
a valid local number and is never stripped.
"""
digits = re.sub(r"\D", "", raw or "")
if len(digits) == 10 and digits.startswith("960"):
digits = digits[3:]
if len(digits) == 7 and digits[0] in ("7", "9"):
return digits
raise ValueError(_PHONE_ERROR)
class DeviceOut(BaseModel):
mac_address: str
group: str | None = None # radusergroup.groupname
status: str | None = None # customers.status
name: str | None = None # customers.name (human metadata)
phone: str | None = None # customers.phone (human metadata)
alias: str | None = None # customers.device_alias (human metadata)
class DeviceCreate(BaseModel):
mac_address: str = Field(pattern=_MAC_RE, max_length=17)
group: str = Field(min_length=1, max_length=64)
name: str = Field(min_length=1, max_length=128, description="Customer name (metadata, ignored by RADIUS)")
phone: str = Field(min_length=1, max_length=32, description="7-digit mobile (9/7…), 960/+960 optional")
alias: str | None = Field(default=None, max_length=64, description="Optional device alias (metadata)")
@field_validator("mac_address")
@classmethod
def _norm(cls, v: str) -> str:
return _normalize_mac(v)
@field_validator("phone")
@classmethod
def _norm_phone(cls, v: str) -> str:
return _normalize_phone(v)
class DeviceEdit(BaseModel):
mac_address: str = Field(pattern=_MAC_RE, max_length=17)
group: str | None = Field(default=None, max_length=64)
status: Literal["new", "paid", "unpaid"] | None = None
name: str | None = Field(default=None, max_length=128)
phone: str | None = Field(default=None, max_length=32)
alias: str | None = Field(default=None, max_length=64)
@field_validator("mac_address")
@classmethod
def _norm(cls, v: str) -> str:
return _normalize_mac(v)
@field_validator("phone")
@classmethod
def _norm_phone(cls, v: str | None) -> str | None:
return None if v is None else _normalize_phone(v)
@model_validator(mode="after")
def _at_least_one(self):
if all(v is None for v in (self.group, self.status, self.name, self.phone, self.alias)):
raise ValueError("provide at least one of: group, status, name, phone, alias")
return self
# ---------- radusergroup ----------
class UserGroupBase(BaseModel):
username: str = Field(max_length=64)
groupname: str = Field(max_length=64)
priority: int = 1
class UserGroupCreate(UserGroupBase):
pass
class UserGroupUpdate(BaseModel):
username: str | None = Field(default=None, max_length=64)
groupname: str | None = Field(default=None, max_length=64)
priority: int | None = None
class UserGroupOut(ORMModel, UserGroupBase):
id: int
# ---------- read-only: radacct ----------
class RadAcctOut(ORMModel):
radacctid: int
acctsessionid: str
acctuniqueid: str
username: str
realm: str | None = None
nasipaddress: str
nasportid: str | None = None
nasporttype: str | None = None
acctstarttime: datetime | None = None
acctupdatetime: datetime | None = None
acctstoptime: datetime | None = None
acctinterval: int | None = None
acctsessiontime: int | None = None
acctauthentic: str | None = None
connectinfo_start: str | None = None
connectinfo_stop: str | None = None
acctinputoctets: int | None = None
acctoutputoctets: int | None = None
calledstationid: str
callingstationid: str
acctterminatecause: str
servicetype: str | None = None
framedprotocol: str | None = None
framedipaddress: str
class_: str | None = Field(default=None, alias="class")
# ---------- read-only: radpostauth ----------
class RadPostAuthOut(ORMModel):
id: int
username: str
reply: str
authdate: datetime | None = None
class_: str | None = Field(default=None, alias="class")
# ---------- read-only: nasreload ----------
class NasReloadOut(ORMModel):
nasipaddress: str
reloadtime: datetime