mgrotke/linuxrouter
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Development

Browse source
This commit is contained in:
Matthew Grotke 2026-06-06 23:22:02 -04:00
parent 8303eb5397
commit a4eb431f22
11 changed files with 744 additions and 1 deletions
Download patch file Download diff file Expand all files Collapse all files

1
docker/routlin-dash/app/config_utils.py
View file

@ -14,6 +14,7 @@ DASHBOARD_LAST_RUN = f'{CONFIGS_DIR}/.dashboard-last-run'
DASHBOARD_LOCK = f'{CONFIGS_DIR}/.dashboard-lock'
DASHBOARD_PENDING = f'{CONFIGS_DIR}/.dashboard-pending'
DASHBOARD_DB = f'{CONFIGS_DIR}/.dashboard-snapshots'
CREDENTIALS_DB = f'{CONFIGS_DIR}/.client-credentials'
HEALTH_FILE = f'{CONFIGS_DIR}/.health'
BLOCKLISTS_DIR = f'{CONFIGS_DIR}/blocklists'
PRODUCT_NAME = os.environ.get('PRODUCT_NAME', 'routlin')

2
docker/routlin-dash/app/main.py
View file

@ -30,6 +30,7 @@ from pages.accountcreate.action import bp as accountcreate_bp
from pages.accountmanage.action import bp as accountmanage_bp
from pages.mdns.action import bp as mdns_bp
from pages.radius.action import bp as radius_bp
from pages.clientcredentials.action import bp as clientcredentials_bp
from action_accountlogout import bp as accountlogout_bp
from api_apply_health import bp as api_apply_health_bp
@ -142,6 +143,7 @@ app.register_blueprint(accountmanage_bp)
app.register_blueprint(accountlogout_bp)
app.register_blueprint(mdns_bp)
app.register_blueprint(radius_bp)
app.register_blueprint(clientcredentials_bp)
app.register_blueprint(api_apply_health_bp)

3
docker/routlin-dash/app/navbar.json
View file

@ -24,7 +24,8 @@
{ "type": "nav_item", "label": "Host Overrides", "map_to": "hostoverrides", "client_requirement": "client_is_administrator+" },
{ "type": "nav_item", "label": "VPN", "map_to": "vpn" },
{ "type": "nav_item", "label": "Banned IPs", "map_to": "bannedips", "client_requirement": "client_is_administrator+" },
{ "type": "nav_item", "label": "RADIUS", "map_to": "radius", "client_requirement": "client_is_administrator+" }
{ "type": "nav_item", "label": "RADIUS", "map_to": "radius", "client_requirement": "client_is_administrator+" },
{ "type": "nav_item", "label": "Client Credentials", "map_to": "clientcredentials", "client_requirement": "client_is_administrator+" }
]
},
{

0
docker/routlin-dash/app/pages/clientcredentials/__init__.py Normal file
View file

301
docker/routlin-dash/app/pages/clientcredentials/action.py Normal file
View file

@ -0,0 +1,301 @@
import hashlib
import sqlite3
import time
from pathlib import Path
import bcrypt
from flask import Blueprint, request, redirect, flash
from auth import require_level
from config_utils import CREDENTIALS_DB
import sanitize
import settings as settings
_PAGE = 'clientcredentials'
PRO_LICENSE = settings.is_pro()
bp = Blueprint(_PAGE, __name__)
# Enum values stored in the database
USER_TYPE_CAPTIVE = 0
USER_TYPE_SUPPLICANT = 1
HASH_CLEARTEXT = 0
HASH_NT = 1
HASH_BCRYPT = 2
VALID_USER_TYPES = {USER_TYPE_CAPTIVE, USER_TYPE_SUPPLICANT}
VALID_HASH_TYPES = {HASH_CLEARTEXT, HASH_NT, HASH_BCRYPT}
# Compatible hash types per user type
COMPATIBLE_HASHES = {
USER_TYPE_CAPTIVE: {HASH_CLEARTEXT, HASH_BCRYPT},
USER_TYPE_SUPPLICANT: {HASH_CLEARTEXT, HASH_NT},
}
# ===================================================================
# DB helpers
# ===================================================================
def _db_conn():
conn = sqlite3.connect(CREDENTIALS_DB)
conn.row_factory = sqlite3.Row
conn.execute("PRAGMA foreign_keys = ON")
conn.execute("""
CREATE TABLE IF NOT EXISTS credentials (
id INTEGER PRIMARY KEY AUTOINCREMENT,
username TEXT NOT NULL UNIQUE COLLATE NOCASE,
password TEXT NOT NULL,
description TEXT NOT NULL DEFAULT '',
user_type INTEGER NOT NULL,
hash_type INTEGER NOT NULL,
vlan TEXT NOT NULL DEFAULT '',
enabled INTEGER NOT NULL DEFAULT 1,
date_set INTEGER NOT NULL,
valid_for INTEGER DEFAULT NULL
)
""")
conn.execute("""
CREATE TABLE IF NOT EXISTS sessions (
id INTEGER PRIMARY KEY AUTOINCREMENT,
ip TEXT NOT NULL,
credential_id INTEGER REFERENCES credentials(id) ON DELETE CASCADE,
started_at INTEGER NOT NULL,
expires_at INTEGER,
mac TEXT NOT NULL DEFAULT ''
)
""")
conn.commit()
return conn
def _get_by_index(conn, row_index):
rows = conn.execute("SELECT * FROM credentials ORDER BY id").fetchall()
if row_index < 0 or row_index >= len(rows):
return None
return rows[row_index]
def _hash_password(plaintext, hash_type):
if hash_type == HASH_CLEARTEXT:
return plaintext
if hash_type == HASH_NT:
try:
return hashlib.new('md4', plaintext.encode('utf-16-le')).hexdigest()
except ValueError:
raise ValueError("NT-Password requires MD4 support. It may be disabled on this system's OpenSSL build.")
if hash_type == HASH_BCRYPT:
return bcrypt.hashpw(plaintext.encode(), bcrypt.gensalt()).decode()
raise ValueError(f"Unknown hash_type: {hash_type}")
def _parse_valid_for(value_str, unit_str):
"""Return valid_for in seconds (int) or None for no expiry."""
unit_str = (unit_str or '').strip()
if unit_str == 'never' or not value_str or not value_str.strip():
return None
try:
n = int(value_str.strip())
if n < 1:
return None
except (ValueError, TypeError):
return None
multipliers = {'hours': 3600, 'days': 86400}
mult = multipliers.get(unit_str)
if mult is None:
return None
return n * mult
def _valid_for_to_display(valid_for):
"""Return (value_str, unit_str) for form pre-population."""
if valid_for is None:
return '', 'never'
if valid_for % 86400 == 0:
return str(valid_for // 86400), 'days'
if valid_for % 3600 == 0:
return str(valid_for // 3600), 'hours'
return str(valid_for // 3600 or 1), 'hours'
def _row_index():
try:
return int(request.form.get('row_index', ''))
except (ValueError, TypeError):
return None
# ===================================================================
# Routes
# ===================================================================
@bp.route('/action/clientcredentials/addedit', methods=['POST'])
@require_level('administrator')
def addedit():
if not PRO_LICENSE:
flash('Client Credentials requires a Routlin Pro license.', 'error')
return redirect(f'/{_PAGE}')
ri = _row_index()
is_edit = ri is not None
username = sanitize.name(request.form.get('username', ''))
password = request.form.get('password', '').strip()
description = sanitize.description(request.form.get('description', ''))
try:
user_type = int(request.form.get('user_type', ''))
hash_type = int(request.form.get('hash_type', ''))
except (ValueError, TypeError):
flash('Invalid user type or hash type.', 'error')
return redirect(f'/{_PAGE}')
if user_type not in VALID_USER_TYPES:
flash('Invalid user type.', 'error')
return redirect(f'/{_PAGE}')
if hash_type not in VALID_HASH_TYPES:
flash('Invalid hash type.', 'error')
return redirect(f'/{_PAGE}')
if hash_type not in COMPATIBLE_HASHES[user_type]:
flash('Selected hash type is not compatible with the selected user type.', 'error')
return redirect(f'/{_PAGE}')
vlan = sanitize.name(request.form.get('vlan', '')) if user_type == USER_TYPE_SUPPLICANT else ''
if user_type == USER_TYPE_SUPPLICANT and not vlan:
flash('VLAN is required for 802.1X supplicant credentials.', 'error')
return redirect(f'/{_PAGE}')
enabled = 'enabled' in request.form
valid_for = _parse_valid_for(
request.form.get('valid_for_value', ''),
request.form.get('valid_for_unit', 'never'),
)
if not username:
flash('Username is required.', 'error')
return redirect(f'/{_PAGE}')
conn = _db_conn()
if is_edit:
existing = _get_by_index(conn, ri)
if not existing:
conn.close()
flash('Credential not found.', 'error')
return redirect(f'/{_PAGE}')
if password:
try:
hashed = _hash_password(password, hash_type)
except ValueError as exc:
conn.close()
flash(str(exc), 'error')
return redirect(f'/{_PAGE}')
stored_password = hashed
stored_hash_type = hash_type
date_set = int(time.time())
else:
stored_password = existing['password']
stored_hash_type = existing['hash_type']
date_set = existing['date_set']
try:
conn.execute(
"""UPDATE credentials
SET username=?, password=?, description=?, user_type=?, hash_type=?,
vlan=?, enabled=?, date_set=?, valid_for=?
WHERE id=?""",
(username, stored_password, description, user_type, stored_hash_type,
vlan, int(enabled), date_set, valid_for, existing['id']),
)
conn.commit()
except sqlite3.IntegrityError:
conn.close()
flash(f"Username '{username}' is already taken.", 'error')
return redirect(f'/{_PAGE}')
conn.close()
flash(f"Credential '{username}' updated.", 'success')
else:
if not password:
flash('Password is required when adding a credential.', 'error')
conn.close()
return redirect(f'/{_PAGE}')
try:
hashed = _hash_password(password, hash_type)
except ValueError as exc:
conn.close()
flash(str(exc), 'error')
return redirect(f'/{_PAGE}')
try:
conn.execute(
"""INSERT INTO credentials
(username, password, description, user_type, hash_type, vlan, enabled, date_set, valid_for)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)""",
(username, hashed, description, user_type, hash_type,
vlan, int(enabled), int(time.time()), valid_for),
)
conn.commit()
except sqlite3.IntegrityError:
conn.close()
flash(f"Username '{username}' already exists.", 'error')
return redirect(f'/{_PAGE}')
conn.close()
flash(f"Credential '{username}' added.", 'success')
return redirect(f'/{_PAGE}')
@bp.route('/action/clientcredentials/delete', methods=['POST'])
@require_level('administrator')
def delete():
if not PRO_LICENSE:
flash('Client Credentials requires a Routlin Pro license.', 'error')
return redirect(f'/{_PAGE}')
ri = _row_index()
if ri is None:
flash('Invalid request.', 'error')
return redirect(f'/{_PAGE}')
conn = _db_conn()
row = _get_by_index(conn, ri)
if not row:
conn.close()
flash('Credential not found.', 'error')
return redirect(f'/{_PAGE}')
conn.execute("DELETE FROM credentials WHERE id=?", (row['id'],))
conn.commit()
conn.close()
flash(f"Credential '{row['username']}' deleted.", 'success')
return redirect(f'/{_PAGE}')
@bp.route('/action/clientcredentials/toggle', methods=['POST'])
@require_level('administrator')
def toggle():
if not PRO_LICENSE:
flash('Client Credentials requires a Routlin Pro license.', 'error')
return redirect(f'/{_PAGE}')
ri = _row_index()
if ri is None:
flash('Invalid request.', 'error')
return redirect(f'/{_PAGE}')
conn = _db_conn()
row = _get_by_index(conn, ri)
if not row:
conn.close()
flash('Credential not found.', 'error')
return redirect(f'/{_PAGE}')
new_enabled = 0 if row['enabled'] else 1
conn.execute("UPDATE credentials SET enabled=? WHERE id=?", (new_enabled, row['id']))
conn.commit()
conn.close()
state = 'enabled' if new_enabled else 'disabled'
flash(f"Credential '{row['username']}' {state}.", 'success')
return redirect(f'/{_PAGE}')

213
docker/routlin-dash/app/pages/clientcredentials/content.json Normal file
View file

@ -0,0 +1,213 @@
{
"client_requirement": "client_is_viewer+",
"items": [
{
"type": "header_page_title",
"items": [
{
"type": "h1",
"text": "Client Credentials"
},
{
"type": "p",
"text": "Username and password credentials for 802.1X supplicants and captive portal authentication."
}
]
},
{
"type": "raw_html",
"html": "%PRO_NOTE%"
},
{
"type": "table",
"datasource": "sqlite:client_credentials",
"empty_message": "No credentials configured.",
"columns": [
{
"label": "Username",
"field": "username",
"class": "col-mono"
},
{
"label": "Description",
"field": "description"
},
{
"label": "Type",
"field": "user_type_label",
"class": "col-narrow"
},
{
"label": "Hash",
"field": "hash_type_label",
"class": "col-narrow"
},
{
"label": "VLAN",
"field": "vlan",
"class": "col-narrow col-mono"
},
{
"label": "Enabled",
"field": "enabled",
"class": "col-narrow",
"render": "badge_toggle",
"toggle_action": "/action/clientcredentials/toggle",
"client_requirement": "client_is_administrator+"
},
{
"label": "Expires",
"field": "expires_label",
"class": "col-narrow"
}
],
"row_actions": [
{
"client_requirement": "client_is_administrator+",
"method": "js_edit",
"target": "add-form",
"text": "Edit",
"class": "btn-ghost btn-sm"
},
{
"client_requirement": "client_is_administrator+",
"action": "/action/clientcredentials/delete",
"method": "post",
"text": "Delete",
"class": "btn-danger btn-sm"
}
]
},
{
"type": "card",
"id": "add-form",
"label": "Add Credential",
"client_requirement": "client_is_administrator+",
"items": [
{
"type": "form",
"action": "/action/clientcredentials/addedit",
"method": "post",
"items": [
{
"type": "hidden",
"name": "row_index",
"value": ""
},
{
"type": "field_row",
"cols": 2,
"items": [
{
"type": "field",
"label": "Username",
"name": "username",
"input_type": "text",
"validate": "VALIDATION_DASH_NAME",
"hint": "Lowercase letters, digits, and hyphens."
},
{
"type": "field",
"label": "Password",
"name": "password",
"input_type": "password",
"hint": "Leave blank when editing to keep the existing password. Changing hash type also requires a new password."
}
]
},
{
"type": "field",
"label": "Description",
"name": "description",
"input_type": "text",
"hint": "Optional label."
},
{
"type": "hr"
},
{
"type": "field_row",
"cols": 2,
"items": [
{
"type": "field",
"label": "User Type",
"name": "user_type",
"input_type": "select",
"options": [
{"value": "0", "label": "Captive Portal"},
{"value": "1", "label": "802.1X Supplicant"}
]
},
{
"type": "field",
"label": "Hash Type",
"name": "hash_type",
"input_type": "select",
"options": []
}
]
},
{
"type": "field",
"label": "VLAN",
"name": "vlan",
"input_type": "select",
"options": "%VLAN_OPTIONS%",
"hint": "VLAN to assign after 802.1X authentication."
},
{
"type": "hr"
},
{
"type": "field",
"label": "Enabled",
"name": "enabled",
"input_type": "checkbox"
},
{
"type": "field_row",
"cols": 2,
"items": [
{
"type": "field",
"label": "Valid For",
"name": "valid_for_value",
"input_type": "number",
"min": 1,
"hint": "How long this credential is valid after creation. Leave blank for no expiry."
},
{
"type": "field",
"label": "Unit",
"name": "valid_for_unit",
"input_type": "select",
"options": [
{"value": "never", "label": "Never (no expiry)"},
{"value": "hours", "label": "Hours"},
{"value": "days", "label": "Days"}
]
}
]
},
{
"type": "button_row",
"items": [
{
"type": "button_primary",
"text": "Add Credential",
"class": "add-credential-btn",
"disabled": "%ADD_CREDENTIAL_DISABLED%"
},
{
"type": "button_cancel",
"text": "Cancel"
}
]
}
]
}
]
}
]
}

96
docker/routlin-dash/app/pages/clientcredentials/view.py Normal file
View file

@ -0,0 +1,96 @@
import json
import sqlite3
import time
import datetime
from config_utils import collect_layout_tokens, CREDENTIALS_DB
from factory import load_json, build_table, table_token_key, iter_table_items, PAGES_DIR
import settings as settings
PRO_LICENSE = settings.is_pro()
USER_TYPE_LABELS = {0: 'Captive Portal', 1: '802.1X'}
HASH_TYPE_LABELS = {0: 'Cleartext', 1: 'NT-Password', 2: 'Bcrypt'}
def _load_credentials():
try:
conn = sqlite3.connect(CREDENTIALS_DB)
conn.row_factory = sqlite3.Row
conn.execute("""
CREATE TABLE IF NOT EXISTS credentials (
id INTEGER PRIMARY KEY AUTOINCREMENT,
username TEXT NOT NULL UNIQUE COLLATE NOCASE,
password TEXT NOT NULL,
description TEXT NOT NULL DEFAULT '',
user_type INTEGER NOT NULL,
hash_type INTEGER NOT NULL,
vlan TEXT NOT NULL DEFAULT '',
enabled INTEGER NOT NULL DEFAULT 1,
date_set INTEGER NOT NULL,
valid_for INTEGER DEFAULT NULL
)
""")
conn.execute("""
CREATE TABLE IF NOT EXISTS sessions (
id INTEGER PRIMARY KEY AUTOINCREMENT,
ip TEXT NOT NULL,
credential_id INTEGER REFERENCES credentials(id) ON DELETE CASCADE,
started_at INTEGER NOT NULL,
expires_at INTEGER,
mac TEXT NOT NULL DEFAULT ''
)
""")
conn.commit()
rows = conn.execute("SELECT * FROM credentials ORDER BY id").fetchall()
conn.close()
return rows
except Exception:
return []
def _format_expiry(date_set, valid_for):
if valid_for is None:
return 'Never'
expires_ts = date_set + valid_for
now = int(time.time())
if expires_ts <= now:
return 'Expired'
dt = datetime.datetime.fromtimestamp(expires_ts)
return dt.strftime('%Y-%m-%d %H:%M')
def collect_tokens(cfg):
tokens = collect_layout_tokens(cfg)
tokens['PRO_NOTE'] = (
'' if PRO_LICENSE else
'<div class="info-bar info-bar-info">'
'<span>Client Credentials is a Routlin Pro feature. '
'Credentials can be viewed but not added or edited without a Pro license.</span></div>'
)
tokens['ADD_CREDENTIAL_DISABLED'] = '' if PRO_LICENSE else 'true'
vlans = [v for v in cfg.get('vlans', []) if not v.get('is_vpn')]
tokens['VLAN_OPTIONS'] = json.dumps(
[{'value': '', 'label': '— Select VLAN —'}] +
[{'value': v['name'], 'label': f"{v['name']} (VLAN {v['vlan_id']})"} for v in vlans]
)
raw_rows = _load_credentials()
display_rows = []
for row in raw_rows:
r = dict(row)
r.pop('password', None)
r['user_type_label'] = USER_TYPE_LABELS.get(r.get('user_type'), str(r.get('user_type', '')))
r['hash_type_label'] = HASH_TYPE_LABELS.get(r.get('hash_type'), str(r.get('hash_type', '')))
r['expires_label'] = _format_expiry(r.get('date_set', 0), r.get('valid_for'))
display_rows.append(r)
content = load_json(f'{PAGES_DIR}/clientcredentials/content.json')
for table_item in iter_table_items(content.get('items', [])):
ds = table_item.get('datasource', '')
data = display_rows if ds == 'sqlite:client_credentials' else []
tokens[table_token_key(ds)] = build_table(table_item, tokens, data)
return tokens