stupid-simple-network-inventory/docs/backend.md

Backend Reference

Entry Point — main.py

FastAPI application. Four router groups protected by require_password_changed (which implies get_current_user + rejects users with must_change_password=True).

app.include_router(auth_router,    prefix="/api/auth",      tags=["auth"])
app.include_router(vlans.router,   prefix="/api/vlans",     dependencies=[Depends(require_password_changed)])
app.include_router(devices.router, prefix="/api/devices",   dependencies=[Depends(require_password_changed)])
app.include_router(discovery.router, prefix="/api/discovery", dependencies=[Depends(require_password_changed)])

GET /api/health returns {"status": "ok"} — public, intentionally minimal (no sensitive detail).

CORS: controlled by the ALLOWED_ORIGINS environment variable (default "*"). Set to your domain in production or leave empty to disable CORS headers (same-origin via proxy).

---

API Endpoints

Auth — /api/auth

POST /api/auth/login

• Auth: None
• Content-Type: application/x-www-form-urlencoded
• Body: username=<str>&password=<str>
• Response: { access_token: str, token_type: "bearer", username: str, must_change_password: bool }
• Errors: 401 if credentials invalid, 429 if rate-limited (IP or username)

PUT /api/auth/account

• Auth: Bearer token (works even when must_change_password=True)
• Body (JSON): { current_password: str, new_username?: str, new_password?: str }
• Response: { access_token: str, token_type: "bearer", username: str, must_change_password: bool } (new token)
• Errors: 400 "Current password is incorrect", 400 "Username already taken", 400 "password_too_short", 400 "password_too_weak", 400 "username_invalid"
• Note: changing the password bumps token_version, invalidating all previously issued tokens.

GET /api/auth/me

• Auth: Bearer token
• Response: { username: str, must_change_password: bool }

---

VLANs — /api/vlans

All endpoints require auth.

GET /api/vlans/

Returns all networks sorted: LANs first (NULL vlan_id), then VLANs ascending by vlan_id.

[{ "id": 1, "vlan_id": null, "name": "LAN", "cidr": "192.168.1.0/24", "color": "#4A90D9" }]

POST /api/vlans/

Creates a network. vlan_id is optional — omit for a plain LAN.

PUT /api/vlans/{id}

Full replacement of a network record.

DELETE /api/vlans/{id}

Deletes the network. SQLAlchemy cascades: associated DeviceInterface rows will lose their vlan_id FK (set to NULL, since the column is nullable).

---

Devices — /api/devices

All endpoints require auth.

GET /api/devices/

Returns all devices with their interfaces, ordered by name.

[{
  "id": 1,
  "name": "proxmox-01",
  "type": "server",
  "description": "Proxmox hypervisor",
  "is_gateway": false,
  "is_livebox": false,
  "virt_type": "baremetal",
  "url": "https://proxmox.local:8006",
  "interfaces": [
    { "id": 1, "device_id": 1, "vlan_id": 2, "ip_address": "192.168.10.5", "name": "eth0", "is_upstream": false }
  ]
}]

POST /api/devices/

Creates device with its interfaces in a single transaction (flush + batch insert + commit).

PUT /api/devices/{id}

Replaces the device and its interfaces: existing interfaces are deleted, new ones inserted.

DELETE /api/devices/{id}

Deletes device and its interfaces (cascade via SQLAlchemy delete-orphan).

Important: when adding a new field to Device, update all four of: models.py, _migrate_* in main.py, DeviceCreate/DeviceOut in routers/devices.py, and the explicit assignments in create_device / update_device. Do not rely on **model_dump() to populate the ORM object — it will silently drop new fields.

---

Discovery — /api/discovery

All endpoints require auth.

POST /api/discovery/ping

Parallel ICMP ping of a list of IPs, 50 concurrent workers.

// Request
{ "ips": ["192.168.1.1", "192.168.1.2"] }

// Response
[{ "ip": "192.168.1.1", "alive": true }, { "ip": "192.168.1.2", "alive": false }]

Uses subprocess.run(["ping", "-c", "1", "-W", "1", ip]). Requires cap_add: NET_RAW on the container (already configured in docker-compose.yml).

POST /api/discovery/scan

Ping sweep + DNS PTR reverse lookup for one or more CIDR ranges.

// Request
{
  "dns_server": "192.168.1.1",
  "targets": [
    { "vlan_id": 1, "cidr": "192.168.1.0/24" }
  ]
}

// Response
{
  "hosts": [{ "ip": "192.168.1.5", "hostname": "nas.local", "vlan_id": 1, "cidr": "192.168.1.0/24" }],
  "total_scanned": 254,
  "duration_s": 3.2
}

Maximum 1024 hosts per target (rejects /21 and wider). 100 concurrent workers. DNS queries use dnspython with a 1s timeout.

---

Database — database.py

engine = create_engine("sqlite:///./data/topology.db", connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

def get_db():          # FastAPI dependency
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

check_same_thread=False is required for SQLite under a multi-threaded ASGI server. The data/ directory is created at module import time.

---

Models — models.py

User

class User(Base):
    __tablename__ = "users"
    id                   = Column(Integer, primary_key=True)
    username             = Column(String, unique=True, nullable=False)
    hashed_password      = Column(String, nullable=False)
    must_change_password = Column(Boolean, nullable=False, default=False, server_default="0")
    token_version        = Column(Integer, nullable=False, default=1, server_default="1")

must_change_password: set to True when admin is bootstrapped with the default password. Clears to False on successful PUT /api/auth/account with a new password.

token_version: incremented on every password change. Included in JWT payload as ver. A token is rejected if its ver differs from the current token_version.

Vlan

class Vlan(Base):
    __tablename__ = "vlans"
    id      = Column(Integer, primary_key=True)
    vlan_id = Column(Integer, unique=True, nullable=True)   # NULL = plain LAN
    name    = Column(String, nullable=False)
    cidr    = Column(String, nullable=True, default="")
    color   = Column(String, default="#4A90D9")
    interfaces = relationship("DeviceInterface", back_populates="vlan")

Device

class Device(Base):
    __tablename__ = "devices"
    id          = Column(Integer, primary_key=True)
    name        = Column(String, nullable=False)
    type        = Column(String, default="other")
    description = Column(String, default="")
    is_gateway  = Column(Boolean, default=False)
    is_livebox  = Column(Boolean, default=False)
    virt_type   = Column(String, nullable=True)   # null | baremetal | lxc | qemu
    url         = Column(String, nullable=True)
    interfaces  = relationship("DeviceInterface", back_populates="device", cascade="all, delete-orphan")

DeviceInterface

class DeviceInterface(Base):
    __tablename__ = "device_interfaces"
    id         = Column(Integer, primary_key=True)
    device_id  = Column(Integer, ForeignKey("devices.id"), nullable=False)
    vlan_id    = Column(Integer, ForeignKey("vlans.id"), nullable=True)
    ip_address = Column(String, nullable=True, default="")
    name       = Column(String, default="eth0")
    is_upstream = Column(Boolean, default=False)

---

Auth — routers/auth.py

Key details for maintainers:

• Secret key: loaded from data/secret_key.txt (auto-generated with secrets.token_hex(32), permissions 0600) or SECRET_KEY env var. Same Docker volume as the database — survives restarts. See README for rotation procedure.
• Token: HS256 JWT, payload { sub: username, ver: token_version, exp: utcnow + 24h }.
• Token invalidation: changing the password increments User.token_version. get_current_user rejects any token whose ver differs from the DB value → immediate invalidation of all old sessions on password change.
• Password hashing: passlib.CryptContext with bcrypt. requirements.txt pins bcrypt==3.2.2 because passlib 1.7.4 is incompatible with bcrypt >= 4.0.
• get_current_user: validates JWT + token version. Returns the User ORM object, or raises 401.
• require_password_changed: wraps get_current_user, raises 403 if must_change_password=True. Used as the dependency for all business routers.
• Rate limiting: 20 attempts/IP/60s and 10 attempts/username/15min (in-memory). Complemented by Nginx limit_req (10 req/min per IP, burst 5) on the login endpoint.

---

Migrations — main.py

All migration functions follow this idempotent pattern:

def _migrate_example():
    with engine.connect() as conn:
        # 1. Bail out if table doesn't exist yet
        if not conn.execute(text("SELECT name FROM sqlite_master WHERE type='table' AND name='table_name'")).fetchone():
            return
        # 2. Check if work is needed
        cols = [row[1] for row in conn.execute(text("PRAGMA table_info(table_name)")).fetchall()]
        if 'new_column' not in cols:
            conn.execute(text("ALTER TABLE table_name ADD COLUMN new_column VARCHAR"))
            conn.commit()

For table recreation (when SQLite cannot ALTER COLUMN), the pattern is:

1. Create table_new with the desired schema
2. INSERT INTO table_new SELECT ... FROM table
3. DROP TABLE table
4. ALTER TABLE table_new RENAME TO table
5. Wrap in PRAGMA foreign_keys=OFF/ON

Always call conn.commit() explicitly — SQLAlchemy 2.0 does not auto-commit DDL on SQLite.

The _migrate_users() function has a two-phase check: table existence, then COUNT(*) = 0 for seeding. This handles the edge case where create_all created the table empty before the migration ran.

_migrate_force_admin_password_change() is the "catch-up" migration for existing installations: if the admin account uses the default bootstrap password ("admin") and must_change_password=0, it sets must_change_password=1. This runs at every startup and is idempotent.

Startup call order (as of phase 3):

1. _migrate_vlan_nullable()
2. _migrate_device_virt_type()
3. _migrate_device_url()
4. _migrate_users_must_change_password()
5. _migrate_users_token_version()
6. _migrate_force_admin_password_change()
7. _migrate_drop_links_table() — drops the links table if it exists (feature removed in phase 3)
8. _migrate_users()
9. Base.metadata.create_all(bind=engine)