Задача - сделать авторизацию клиентов в гостевой сети по коду из sms. Схема подключени клиента к гостевой сети будет примерно такой: подключение клиента к wifi введя пароль, CaptivePortal на keenetic автоматически перенаправляет клиента на страницу UAM сервера где нужно ввести свой номер телефона и подтвердить его кодом из смс, при верном коде UAM сервер заводит пользователя в поле имя которого пишет номер телефона, далее keenetic проверяет разрешен ли доступ в radius, при разрешении выдается доступ к wifi на 15 минут.
1. Первый этап. Нам нужно настроить отправку смс клиентам.
Для этого используется модем Huawei E1550. Так как мой сервер ubuntu на виртуалке будем продрасывать модем туда по usbip. Общяа интрукция проброса тут:
Настройка отправки смс с помощью gammu
2. Второй этап установим freeradius на ubuntu
Задаём статик IP Ubuntu
network:
version: 2
ethernets:
eth0:
addresses:
- "192.168.7.10/24"
nameservers:
addresses:
- 192.168.7.1
search: []
routes:
- to: "default"
via: "192.168.7.1"
Устанавливаем пакеты
Проверяем:
пакет FreeRADIUS сейчас находится в ветке 3.2.x
Настраиваем FreeRADIUS
сделаем резервную копию:
добавим в клиенты наш keenetic
В конец добавляем:
client keenetic {
ipaddr = 192.168.7.94
secret = MyRadiusSecret_2026
shortname = keenetic-start
nastype = other
}
IP здесь должен быть адресом Keenetic, с которого он отправляет RADIUS-запросы.
Проверяем конфигурацию
В конце должно быть что-нибудь вроде:
Configuration appears to be OK
Для диагностики FreeRADIUS рекомендует запуск в debug-режиме:
Проверяем RADIUS без Keenetic
В другом SSH/терминале:
По дефолту эта настройка активна
Ожидаем:
Received Access-Accept
3. Третий этап. Установим и настроим MariaDB для хранения всех данных.
Теперь создаём отдельную БД для FreeRADIUS и пользователя в MariaDB.
CREATE DATABASE radius CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;
CREATE USER 'radius'@'localhost' IDENTIFIED BY 'RadiusPass';
GRANT ALL PRIVILEGES ON radius.* TO 'radius'@'localhost';
FLUSH PRIVILEGES;
EXIT;
Теперь проверим подключение именно этим пользователем:
Следующий шаг — установить MYSQL-модуль FreeRADIUS и загрузить штатную схему таблиц
Затем проверим, появился ли модуль:
И посмотрим, какие модули сейчас включены:
Включаем модуль
Теперь настроим SQL-модуль на MariaDB.
Замените параметры:
driver = "rlm_sql_null"dialect = ${modules.sql.dialect}
и замените на:
Далее найдите блок подключения к базе, примерно такой:
server = "localhost"port = 3306login = "radius"password = "..."database = "radius"
Укажите:
Загружаем штатные таблицы FreeRADIUS
dpkg -L freeradius-mysql | grep schema.sql
если ничего не нашло
у нас это
Проверим таблицы:
проверить, что FreeRADIUS действительно может подключиться к MariaDB.
если включен TLS отключаем его, он не нужен
заменить:
mysql { tls { ca_file = "/etc/ssl/certs/my_ca.crt" ca_path = "/etc/ssl/certs/" certificate_file = "/etc/ssl/certs/private/client.crt" private_key_file = "/etc/ssl/certs/private/client.key" cipher = "DHE-RSA-AES256-SHA:AES128-SHA" tls_required = yes tls_check_cert = no tls_check_cert_cn = no } warnings = auto}
На:
Должно быть Configuration appears to be OK
sudo nano /etc/freeradius/3.0/sites-enabled/default
в секции
authorize {
заменить
-sql
на:
sql
sudo freeradius –XC
4. Этап. Создадим наш UAM сервер.
Скрипт будет на Pyton
Через Python-библиотеку pyrad Flask будет отправлять в FreeRADIUS Access-Request только с номером телефона, без User-Password
создадим директорию для flask
chown $USER:$USER /opt/keenetic-uam
cd /opt/keenetic-uam
Создаём virtualenv:
python3 -m venv venv
source venv/bin/activate
Устанавливаем Flask и RADIUS-клиент:
pip install flask
Установим pyrad
проверим
python3 -c "import pyrad; print('pyrad OK')"
#!/usr/bin/env python3
import os
import re
import time
import hmac
import hashlib
import secrets
import sqlite3
import subprocess
import urllib.parse
from flask import Flask, request, redirect, render_template_string
from pyrad.client import Client
from pyrad.dictionary import Dictionary
from pyrad.packet import AccessRequest, AccessAccept, AccessReject
app = Flask(__name__)
# ============================================================
# Основные настройки
# ============================================================
DB_PATH = "/opt/keenetic-uam/otp.db"
GAMMU_CONFIG = "/home/root/gammurc"
# Должен совпадать с UAM Secret на Keenetic
UAM_SECRET = "UAMPASSWORD"
OTP_TTL = 300 # 5 минут
MAX_ATTEMPTS = 5
# ============================================================
# Keenetic UAM
#
# Пароль используется только для текущего UAM /logon CHAP.
# Username пользователя здесь больше не хранится:
# username = номер телефона.
# ============================================================
#KEENETIC_PASSWORD = "R17Bc0mSh"
# ============================================================
# FreeRADIUS
# ============================================================
RADIUS_HOST = "127.0.0.1"
RADIUS_PORT = 1812
# Должен совпадать с secret соответствующего client
# в конфигурации FreeRADIUS.
RADIUS_SECRET = b"MyRadiusSecret_2026"
RADIUS_DICTIONARY = "/opt/keenetic-uam/radius_dictionary"
# ============================================================
# HTML
# ============================================================
HTML = """
<!doctype html>
<html lang="ru">
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<title>Wi-Fi доступ</title>
<style>
* {
box-sizing: border-box;
}
html, body {
margin: 0;
padding: 0;
min-height: 100%;
}
body {
min-height: 100vh;
font-family: Arial, sans-serif;
background-image:
linear-gradient(
rgba(0, 0, 0, 0.35),
rgba(0, 0, 0, 0.35)
),
url("/static/background.png");
background-size: cover;
background-position: center;
background-repeat: no-repeat;
display: flex;
justify-content: center;
align-items: center;
padding: 20px;
}
.login-box {
width: 100%;
max-width: 420px;
background: rgba(255, 255, 255, 0.20);
border: 1px solid rgba(255, 255, 255, 0.35);
border-radius: 18px;
padding: 30px;
box-shadow: 0 10px 40px rgba(0, 0, 0, 0.30);
backdrop-filter: blur(12px);
-webkit-backdrop-filter: blur(12px);
}
h2 {
text-align: center;
margin-top: 0;
}
input,
button {
width: 100%;
padding: 13px;
margin-top: 12px;
font-size: 16px;
border-radius: 8px;
border: 1px solid #ccc;
}
button {
border: none;
background: #1677ff;
color: white;
cursor: pointer;
}
button:hover {
background: #0d5dcc;
}
button:disabled {
opacity: 0.6;
cursor: default;
}
.error {
color: #b00020;
margin: 15px 0;
}
.info {
color: #333;
margin: 15px 0;
}
</style>
</head>
<body>
<div class="login-box">
<h2>Доступ к Wi-Fi</h2>
{% if error %}
<div class="error">{{ error }}</div>
{% endif %}
{% if info %}
<div class="info">{{ info }}</div>
{% endif %}
{{ body|safe }}
</div>
</body>
</html>
"""
# ============================================================
# SQLite
# ============================================================
def db_connect():
connection = sqlite3.connect(DB_PATH)
connection.row_factory = sqlite3.Row
return connection
def init_db():
os.makedirs(os.path.dirname(DB_PATH), exist_ok=True)
with db_connect() as db:
db.execute("""
CREATE TABLE IF NOT EXISTS otp_sessions (
id INTEGER PRIMARY KEY AUTOINCREMENT,
phone TEXT NOT NULL,
otp_hash TEXT NOT NULL,
expires_at INTEGER NOT NULL,
attempts INTEGER NOT NULL DEFAULT 0,
used INTEGER NOT NULL DEFAULT 0,
uamip TEXT,
uamport TEXT,
challenge TEXT,
userurl TEXT,
mac TEXT,
client_ip TEXT,
nasid TEXT,
sessionid TEXT,
created_at INTEGER NOT NULL
)
""")
db.commit()
# ============================================================
# RADIUS dictionary
# ============================================================
def init_radius_dictionary():
os.makedirs(
os.path.dirname(RADIUS_DICTIONARY),
exist_ok=True
)
dictionary_content = """ATTRIBUTE User-Name 1 string
ATTRIBUTE NAS-IP-Address 4 ipaddr
ATTRIBUTE Calling-Station-Id 31 string
"""
try:
with open(
RADIUS_DICTIONARY,
"w",
encoding="utf-8"
) as file:
file.write(dictionary_content)
except OSError as error:
app.logger.error(
"Cannot create RADIUS dictionary: %s",
error
)
raise
# ============================================================
# Phone
# ============================================================
def normalize_phone(phone):
phone = phone.strip()
phone = re.sub(
r"[ \-()]",
"",
phone
)
if phone.startswith("00"):
phone = "+" + phone[2:]
if not re.fullmatch(
r"\+[0-9]{8,15}",
phone
):
return None
return phone
# ============================================================
# OTP
# ============================================================
def hash_otp(otp):
return hashlib.sha256(
otp.encode("utf-8")
).hexdigest()
# ============================================================
# SMS
# ============================================================
def send_sms(phone, message):
app.logger.info(
"Sending SMS to %s",
phone
)
command = [
"gammu",
"--config",
GAMMU_CONFIG,
"sendsms",
"TEXT",
phone,
"-unicode",
"-text",
message
]
result = subprocess.run(
command,
capture_output=True,
text=True,
encoding="utf-8",
errors="replace",
timeout=60
)
if result.returncode != 0:
error_text = (
result.stderr.strip()
or result.stdout.strip()
)
app.logger.error(
"Gammu SMS error: %s",
error_text or "unknown error"
)
raise RuntimeError(
error_text or "Gammu не смог отправить SMS"
)
app.logger.info(
"SMS successfully sent to %s",
phone
)
# ============================================================
# MariaDB / radcheck
#
# Используем стандартную таблицу FreeRADIUS.
#
# Логика:
#
# номера нет -> INSERT Accept
# уже Accept -> ничего
# уже Reject -> ничего
#
# Поэтому правильный SMS никогда автоматически
# не снимает административный запрет.
# ============================================================
def ensure_radius_user(phone):
app.logger.info(
"Checking radcheck entry for %s",
phone
)
# Значение телефона содержит только + и цифры,
# поэтому после normalize_phone оно безопасно
# для этого SQL-запроса.
escaped_phone = phone.replace("'", "''")
select_sql = (
"SELECT value "
"FROM radcheck "
"WHERE username = '"
+ escaped_phone
+ "' "
"AND attribute = 'Auth-Type' "
"ORDER BY id "
"LIMIT 1;"
)
try:
result = subprocess.run(
[
"mariadb",
"--batch",
"--skip-column-names",
"radius",
"-e",
select_sql
],
capture_output=True,
text=True,
encoding="utf-8",
errors="replace",
timeout=10
)
except subprocess.TimeoutExpired:
app.logger.error(
"MariaDB query timeout for %s",
phone
)
return False
if result.returncode != 0:
app.logger.error(
"MariaDB SELECT error: %s",
result.stderr.strip()
)
return False
value = result.stdout.strip()
# --------------------------------------------------------
# Пользователь ещё не существует.
# Добавляем разрешение.
# --------------------------------------------------------
if not value:
insert_sql = (
"INSERT INTO radcheck "
"(username, attribute, op, value) "
"VALUES ('"
+ escaped_phone
+ "', 'Auth-Type', ':=', 'Accept');"
)
try:
result = subprocess.run(
[
"mariadb",
"radius",
"-e",
insert_sql
],
capture_output=True,
text=True,
encoding="utf-8",
errors="replace",
timeout=10
)
except subprocess.TimeoutExpired:
app.logger.error(
"MariaDB INSERT timeout for %s",
phone
)
return False
if result.returncode != 0:
app.logger.error(
"MariaDB INSERT error: %s",
result.stderr.strip()
)
return False
app.logger.info(
"New RADIUS user added: %s -> Accept",
phone
)
return True
# --------------------------------------------------------
# Пользователь уже существует.
# --------------------------------------------------------
if value.lower() == "accept":
app.logger.info(
"RADIUS user %s already enabled",
phone
)
return True
if value.lower() == "reject":
app.logger.warning(
"RADIUS user %s is administratively blocked",
phone
)
# Ничего не меняем!
return True
# Неизвестное значение Auth-Type.
app.logger.error(
"Unexpected Auth-Type value for %s: %s",
phone,
value
)
return False
# ============================================================
# FreeRADIUS
#
# В запросе НЕТ User-Password.
#
# FreeRADIUS проверяет username через radcheck.
# ============================================================
def check_radius(username, calling_station_id=None):
app.logger.info(
"Checking RADIUS authorization for user %s",
username
)
try:
dictionary = Dictionary(
RADIUS_DICTIONARY
)
client = Client(
server=RADIUS_HOST,
secret=RADIUS_SECRET,
dict=dictionary
)
client.authport = RADIUS_PORT
request_packet = client.CreateAuthPacket(
code=AccessRequest,
User_Name=username
)
request_packet["NAS-IP-Address"] = RADIUS_HOST
if calling_station_id:
request_packet["Calling-Station-Id"] = (
calling_station_id
)
response = client.SendPacket(
request_packet
)
if response.code == AccessAccept:
app.logger.info(
"RADIUS Access-Accept for user %s",
username
)
return True
if response.code == AccessReject:
app.logger.warning(
"RADIUS Access-Reject for user %s",
username
)
return False
app.logger.warning(
"Unexpected RADIUS response code %s for %s",
response.code,
username
)
return False
except Exception as error:
app.logger.error(
"RADIUS error for user %s: %s",
username,
error
)
return False
# ============================================================
# Keenetic UAM CHAP
# ============================================================
def calculate_chap_response(
challenge,
password
):
challenge_bytes = bytes.fromhex(
challenge
)
new_challenge = hashlib.md5(
challenge_bytes
+ UAM_SECRET.encode("utf-8")
).digest()
response = hashlib.md5(
b"\x00"
+ password.encode("utf-8")
+ new_challenge
).hexdigest()
return response
def make_keenetic_redirect(session):
challenge = session["challenge"]
response = calculate_chap_response(
challenge,
KEENETIC_PASSWORD
)
# Передаём в CoovaChilli именно номер телефона.
# Это username, который затем попадёт
# в RADIUS и в radacct.
phone = session["phone"]
params = {
"username": phone,
"response": response,
"userurl": (
session["userurl"]
or "http://example.com"
)
}
url = (
f"http://{session['uamip']}:{session['uamport']}/logon?"
+ urllib.parse.urlencode(params)
)
app.logger.info(
"Redirecting authenticated user %s to Keenetic UAM",
phone
)
return redirect(url)
# ============================================================
# /health
# ============================================================
@app.route("/health")
def health():
return "OK", 200
# ============================================================
# /login
# ============================================================
@app.route(
"/login",
methods=["GET"]
)
def login():
if request.args.get("res") == "success":
app.logger.info(
"Keenetic reports successful authentication for user %s",
request.args.get(
"uid",
"unknown"
)
)
return render_template_string(
HTML,
error=None,
info="Доступ к Wi-Fi разрешён.",
body=""
)
if request.args.get("res") == "failed":
app.logger.warning(
"Keenetic reports authentication failure"
)
return render_template_string(
HTML,
error="Авторизация не удалась.",
info=None,
body=""
)
params = {
"uamip": request.args.get(
"uamip",
""
),
"uamport": request.args.get(
"uamport",
""
),
"challenge": request.args.get(
"challenge",
""
),
"userurl": request.args.get(
"userurl",
""
),
"mac": request.args.get(
"mac",
""
),
"ip": request.args.get(
"ip",
""
),
"nasid": request.args.get(
"nasid",
""
),
"sessionid": request.args.get(
"sessionid",
""
)
}
app.logger.info(
"New UAM login request: MAC=%s IP=%s Session=%s",
params["mac"],
params["ip"],
params["sessionid"]
)
body = """
<p>Введите номер телефона для получения кода:</p>
<form
method="post"
action="/send-code"
id="phone-form"
>
<input
type="tel"
name="phone"
placeholder="+79991234567"
required
>
{% for key, value in params.items() %}
<input
type="hidden"
name="{{ key }}"
value="{{ value }}"
>
{% endfor %}
<button
type="submit"
id="send-code-button"
>
Получить код
</button>
</form>
<script>
(function () {
const form = document.getElementById("phone-form");
const button = document.getElementById("send-code-button");
if (!form || !button) {
return;
}
form.addEventListener("submit", function () {
if (button.disabled) {
return;
}
// Блокируем кнопку СРАЗУ при отправке формы,
// до фактического POST /send-code.
button.disabled = true;
button.textContent = "Отправка...";
});
})();
</script>
"""
return render_template_string(
HTML,
error=None,
info=None,
body=render_template_string(
body,
params=params
)
)
# ============================================================
# /send-code
# ============================================================
@app.route(
"/send-code",
methods=["POST"]
)
def send_code():
phone = normalize_phone(
request.form.get(
"phone",
""
)
)
if not phone:
app.logger.warning(
"Invalid phone number submitted"
)
return render_template_string(
HTML,
error=(
"Введите номер в международном формате, "
"например +79991234567."
),
info=None,
body=""
)
otp = str(
secrets.randbelow(900000)
+ 100000
)
now = int(time.time())
with db_connect() as db:
cursor = db.execute(
"""
INSERT INTO otp_sessions (
phone,
otp_hash,
expires_at,
uamip,
uamport,
challenge,
userurl,
mac,
client_ip,
nasid,
sessionid,
created_at
)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
phone,
hash_otp(otp),
now + OTP_TTL,
request.form.get(
"uamip",
""
),
request.form.get(
"uamport",
""
),
request.form.get(
"challenge",
""
),
request.form.get(
"userurl",
""
),
request.form.get(
"mac",
""
),
request.form.get(
"ip",
""
),
request.form.get(
"nasid",
""
),
request.form.get(
"sessionid",
""
),
now
)
)
token = cursor.lastrowid
db.commit()
app.logger.info(
"OTP generated for %s, session token=%s",
phone,
token
)
message = (
f"Wi-Fi код: {otp}. Действует 5 минут."
)
try:
send_sms(
phone,
message
)
except Exception as error:
with db_connect() as db:
db.execute(
"DELETE FROM otp_sessions WHERE id = ?",
(token,)
)
db.commit()
app.logger.error(
"SMS sending failed for session %s: %s",
token,
error
)
return render_template_string(
HTML,
error=f"Ошибка отправки SMS: {error}",
info=None,
body=""
)
body = f"""
<p>Код отправлен на номер <b>{phone}</b>.</p>
<form method="post" action="/verify-code">
<input
type="text"
name="otp"
inputmode="numeric"
pattern="[0-9]{{6}}"
maxlength="6"
placeholder="Введите код"
required
>
<input
type="hidden"
name="token"
value="{token}"
>
<button type="submit">
Подтвердить
</button>
</form>
"""
return render_template_string(
HTML,
error=None,
info=None,
body=body
)
# ============================================================
# /verify-code
# ============================================================
@app.route(
"/verify-code",
methods=["POST"]
)
def verify_code():
token = request.form.get(
"token",
""
)
entered_otp = request.form.get(
"otp",
""
).strip()
app.logger.info(
"OTP verification started, token=%s",
token
)
if (
not token.isdigit()
or not re.fullmatch(
r"[0-9]{6}",
entered_otp
)
):
app.logger.warning(
"Invalid OTP format for token=%s",
token
)
return render_template_string(
HTML,
error="Введите шестизначный код.",
info=None,
body=""
)
with db_connect() as db:
session = db.execute(
"SELECT * FROM otp_sessions WHERE id = ?",
(int(token),)
).fetchone()
if session is None:
app.logger.warning(
"OTP session not found: token=%s",
token
)
return render_template_string(
HTML,
error="Сессия не найдена.",
info=None,
body=""
)
if session["used"]:
app.logger.warning(
"OTP already used: token=%s",
token
)
return render_template_string(
HTML,
error="Этот код уже использован.",
info=None,
body=""
)
if int(time.time()) > session["expires_at"]:
app.logger.warning(
"OTP expired: token=%s",
token
)
return render_template_string(
HTML,
error="Срок действия кода истёк.",
info=None,
body=""
)
if session["attempts"] >= MAX_ATTEMPTS:
app.logger.warning(
"Too many OTP attempts: token=%s",
token
)
return render_template_string(
HTML,
error="Превышено количество попыток.",
info=None,
body=""
)
expected_hash = session["otp_hash"]
actual_hash = hash_otp(
entered_otp
)
if not hmac.compare_digest(
expected_hash,
actual_hash
):
db.execute(
"""
UPDATE otp_sessions
SET attempts = attempts + 1
WHERE id = ?
""",
(int(token),)
)
db.commit()
app.logger.warning(
"Invalid OTP: token=%s attempt=%s",
token,
session["attempts"] + 1
)
return render_template_string(
HTML,
error="Неверный код.",
info=None,
body=""
)
# =====================================================
# OTP успешно проверен.
# Используем именно номер телефона пользователя.
# =====================================================
phone = session["phone"]
app.logger.info(
"OTP verification successful for user %s",
phone
)
# =====================================================
# ВАЖНО: OTP не добавляет номер телефона в radcheck.
# OTP только подтверждает владение номером.
# Разрешён ли номер, решает FreeRADIUS.
# =====================================================
# =====================================================
# Проверяем пользователя через FreeRADIUS.
#
# User-Password отсутствует.
# =====================================================
if not ensure_radius_user(phone):
app.logger.error(
"Unable to authorize RADIUS user %s",
phone
)
return render_template_string(
HTML,
error="Не удалось проверить доступ к Wi-Fi.",
info=None,
body=""
)
radius_ok = check_radius(
phone,
session["mac"]
)
if not radius_ok:
app.logger.warning(
"OTP correct, but RADIUS rejected user %s",
phone
)
return render_template_string(
HTML,
error="Для этого номера доступ к Wi-Fi запрещён.",
info=None,
body=""
)
# =====================================================
# OTP + RADIUS успешно.
# =====================================================
app.logger.info(
"OTP + RADIUS authentication successful for user %s",
phone
)
# Только после успешного RADIUS
# делаем OTP использованным.
db.execute(
"""
UPDATE otp_sessions
SET used = 1
WHERE id = ?
""",
(int(token),)
)
db.commit()
# =========================================================
# RADIUS подтвердил пользователя.
# Отправляем Keenetic успешный UAM logon.
# =========================================================
return make_keenetic_redirect(
session
)
# ============================================================
# Запуск
# ============================================================
if __name__ == "__main__":
init_db()
init_radius_dictionary()
app.run(
host="127.0.0.1",
port=8080,
debug=False
)
Запускаем UAM
cd /opt/keenetic-uam
source venv/bin/activate
python app.py
Должно появиться:
Running on http://0.0.0.0:8080
С другого компьютера:
http://192.168.1.10:8080/login
Делаем systemd-сервис
Description=Keenetic UAM Captive Portal
After=network.target[Service]
User=root
WorkingDirectory=/opt/keenetic-uam
ExecStart=/opt/keenetic-uam/venv/bin/python /opt/keenetic-uam/app.py
Restart=always
RestartSec=3[Install]
WantedBy=multi-user.target
systemctl daemon-reload
systemctl enable --now keenetic-uam
Нужно в каталог /opt/keenetic-uam/static/background.png фоновое избражение .
Проверяем:
5. Пятый этап. Развернем Nginx для web доступа к нашей странице авторизации.
Устанавливаем reverse proxy:
server {
listen 80;
server_name 192.168.1.10;
location / {
proxy_pass http://127.0.0.1:8080;
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
}
}
Активируем:
Проверяем:
nginx –t
systemctl restart nginx
Проверяем
6. Шестой Этап. Создание статистики.
Создадим отдельного пользователя MariaDB
Лучше не давать статистическому приложению права записи:
В MariaDB:
CREATE USER 'radius_stat'@'localhost' IDENTIFIED BY 'PASSSTATISTIC';
GRANT SELECT ON radius.radacct TO 'radius_stat'@'localhost';
FLUSH PRIVILEGES;
EXIT;
Создадим директорию для приложения
mkdir -p /opt/uat/stat
apt install -y python3-venv python3-pip
Создали новое виртуальное окружение:
sudo python3 -m venv /opt/uat/venv
sudo /opt/uat/venv/bin/pip install flask pymysql
nano /opt/uat/app_stat.py
import pymysqlapp = Flask(
__name__,
template_folder="/opt/uat/stat"
)DB_CONFIG = {
"host": "127.0.0.1",
"user": "radius_stat",
"password": "FAsk0^lkH",
"database": "radius",
"charset": "utf8mb4",
"cursorclass": pymysql.cursors.DictCursor,
}def db_connect():
return pymysql.connect(**DB_CONFIG)@app.route("/stats/ap", methods=["GET", "POST"])
def ap_list():
if request.method == "POST":
ap_id = request.form.get("id", "").strip()
name = request.form.get("name", "").strip()
ip = request.form.get("ip", "").strip()
if ap_id:
with db_connect() as conn:
with conn.cursor() as cursor:
cursor.execute(
"""
UPDATE ap
SET name = %s,
ip = %s
WHERE id = %s
""",
(
name,
ip or None,
ap_id
)
)
conn.commit()
with db_connect() as conn:
with conn.cursor() as cursor:
cursor.execute(
"""
SELECT
id,
mac,
name,
ip
FROM ap
ORDER BY mac
"""
)
aps = cursor.fetchall()
return render_template(
"ap.html",
aps=aps
)
@app.route("/stats")
def stats():
phone = request.args.get(
"phone",
""
).strip()
ap = request.args.get(
"ap",
""
).strip()
date_from = request.args.get(
"date_from",
""
).strip()
date_to = request.args.get(
"date_to",
""
).strip()
where = []
params = []
if phone:
where.append(
"radacct.username LIKE %s"
)
params.append(
f"%{phone}%"
)
if ap:
where.append(
"radacct.calledstationid = %s"
)
params.append(
ap
)
if date_from:
where.append(
"radacct.acctstarttime >= %s"
)
params.append(
f"{date_from} 00:00:00"
)
if date_to:
# Включаем весь последний день
where.append(
"radacct.acctstarttime < DATE_ADD(%s, INTERVAL 1 DAY)"
)
params.append(
date_to
)
where_sql = ""
if where:
where_sql = (
"WHERE " +
" AND ".join(where)
)
# Основной запрос статистики
query = f"""
SELECT
radacct.radacctid,
radacct.username,
radacct.calledstationid,
ap.name AS ap_name,
radacct.callingstationid,
radacct.nasipaddress,
radacct.nasportid,
radacct.acctstarttime,
radacct.acctupdatetime,
radacct.acctstoptime,
radacct.acctsessiontime,
radacct.acctinputoctets,
radacct.acctoutputoctets,
radacct.acctterminatecause,
radacct.framedipaddress
FROM radacct
LEFT JOIN ap
ON ap.mac = radacct.calledstationid
{where_sql}
ORDER BY radacct.acctstarttime DESC
LIMIT 1000
"""
with db_connect() as conn:
with conn.cursor() as cursor:
# Получаем сессии
cursor.execute(
query,
params
)
sessions = cursor.fetchall()
# Список AP для фильтра
cursor.execute(
"""
SELECT
radacct.calledstationid,
ap.name AS ap_name
FROM radacct
LEFT JOIN ap
ON ap.mac = radacct.calledstationid
WHERE radacct.calledstationid <> ''
GROUP BY
radacct.calledstationid,
ap.name
ORDER BY
radacct.calledstationid
"""
)
aps = cursor.fetchall()
# Форматирование данных
for row in sessions:
seconds = (
row["acctsessiontime"]
or 0
)
if seconds:
hours = seconds // 3600
minutes = (
seconds % 3600
) // 60
secs = seconds % 60
if hours:
row["duration"] = (
f"{hours} ч "
f"{minutes} мин"
)
elif minutes:
row["duration"] = (
f"{minutes} мин "
f"{secs} сек"
)
else:
row["duration"] = (
f"{secs} сек"
)
else:
row["duration"] = "—"
input_bytes = (
row["acctinputoctets"]
or 0
)
output_bytes = (
row["acctoutputoctets"]
or 0
)
row["input_mb"] = round(
input_bytes / 1024 / 1024,
2
)
row["output_mb"] = round(
output_bytes / 1024 / 1024,
2
)
return render_template(
"stats.html",
sessions=sessions,
aps=aps,
phone=phone,
ap=ap,
date_from=date_from,
date_to=date_to,
)
if __name__ == "__main__":
app.run(
host="127.0.0.1",
port=8081,
debug=False
)
nano /opt/uat/stat/stats.html
<html lang="ru"><head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1"><title>Статистика Wi-Fi</title><style>
* {
box-sizing: border-box;
}body {
margin: 0;
padding: 25px;
font-family: Arial, sans-serif;
background: #f4f6f8;
color: #222;
}
.container {
max-width: 1500px;
margin: auto;
}
h1 {
margin-bottom: 20px;
}
.filters {
background: white;
padding: 20px;
border-radius: 12px;
margin-bottom: 20px;
box-shadow: 0 2px 10px rgba(0, 0, 0, .08);
}
.filters form {
display: flex;
flex-wrap: wrap;
gap: 12px;
align-items: end;
}
.field {
display: flex;
flex-direction: column;
gap: 5px;
}
label {
font-size: 13px;
color: #666;
}
input,
select,
button {
height: 40px;
padding: 0 10px;
border: 1px solid #ccc;
border-radius: 7px;
font-size: 14px;
}
button {
background: #1677ff;
color: white;
border: none;
padding: 0 20px;
cursor: pointer;
}
button:hover {
background: #0d5dcc;
}
.table-wrapper {
background: white;
border-radius: 12px;
overflow-x: auto;
box-shadow: 0 2px 10px rgba(0, 0, 0, .08);
}
table {
width: 100%;
border-collapse: collapse;
min-width: 1100px;
}
th,
td {
padding: 11px 12px;
border-bottom: 1px solid #eee;
text-align: left;
white-space: nowrap;
}
th {
background: #f8f9fa;
font-size: 13px;
}
td {
font-size: 13px;
}
tr:hover {
background: #f8fbff;
}
.empty {
padding: 40px;
text-align: center;
color: #777;
}
.online {
color: #087f23;
font-weight: bold;
}
.timeout {
color: #b35a00;
}
.traffic {
text-align: right;
}
/* Название AP — кликабельная ссылка */
.ap-name {
color: #1677ff;
text-decoration: none;
font-weight: 600;
}
.ap-name:hover {
text-decoration: underline;
}
</style>
</head>
<body>
<div class="container">
<h1>Статистика Wi-Fi</h1>
<div class="filters">
<form method="get" action="/stats">
<!-- Телефон -->
<div class="field">
<label>Телефон</label>
<input
type="text"
name="phone"
value="{{ phone }}"
placeholder="+79595082552"
>
</div>
<!-- AP -->
<div class="field">
<label>AP</label>
<select name="ap">
<option value="">
Все точки
</option>
{% for item in aps %}
<option
value="{{ item.calledstationid }}"
{% if ap == item.calledstationid %}selected{% endif %}
>
{{ item.calledstationid }}
{% if item.ap_name %}
— {{ item.ap_name }}
{% endif %}
</option>
{% endfor %}
</select>
</div>
<!-- Дата начала -->
<div class="field">
<label>С даты</label>
<input
type="date"
name="date_from"
value="{{ date_from }}"
>
</div>
<!-- Дата окончания -->
<div class="field">
<label>По дату</label>
<input
type="date"
name="date_to"
value="{{ date_to }}"
>
</div>
<button type="submit">
Показать
</button>
</form>
</div>
<!-- Таблица -->
<div class="table-wrapper">
{% if sessions %}
<table>
<thead>
<tr>
<th>Начало</th>
<th>Телефон</th>
<th>AP</th>
<th>MAC клиента</th>
<th>IP</th>
<th>Окончание</th>
<th>Длительность</th>
<th>Входящий MB</th>
<th>Исходящий MB</th>
<th>Причина</th>
</tr>
</thead>
<tbody>
{% for row in sessions %}
<tr>
<!-- Начало -->
<td>
{{ row.acctstarttime or "—" }}
</td>
<!-- Телефон -->
<td>
{{ row.username }}
</td>
<!-- AP -->
<td>
{{ row.calledstationid }}
{% if row.ap_name %}
—
<a
href="/stats/ap"
class="ap-name"
>
{{ row.ap_name }}
</a>
{% endif %}
</td>
<!-- MAC клиента -->
<td>
{{ row.callingstationid }}
</td>
<!-- IP клиента -->
<td>
{{ row.framedipaddress }}
</td>
<!-- Окончание -->
<td>
{% if row.acctstoptime %}
{{ row.acctstoptime }}
{% else %}
<span class="online">
онлайн
</span>
{% endif %}
</td>
<!-- Длительность -->
<td>
{{ row.duration }}
</td>
<!-- Входящий трафик -->
<td class="traffic">
{{ row.input_mb }}
</td>
<!-- Исходящий трафик -->
<td class="traffic">
{{ row.output_mb }}
</td>
<!-- Причина завершения -->
<td>
{% if row.acctterminatecause == "Session-Timeout" %}
<span class="timeout">
Session-Timeout
</span>
{% else %}
{{ row.acctterminatecause or "—" }}
{% endif %}
</td>
</tr>
{% endfor %}
</tbody>
</table>
{% else %}
<div class="empty">
Сессий не найдено
</div>
{% endif %}
</div>
</div>
</body>
</html>
Сначала запускаем вручную
Не делаем systemd и nginx, пока не убедимся, что само приложение работает
проверим curl
Добавляем в NGINX
Сделай так:
listen 80;
server_name 192.168.7.10;location /stats {
proxy_pass http://127.0.0.1:8081;
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_set_header X-Forwarded-Proto $scheme;
}location / {
proxy_pass http://127.0.0.1:8080;
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_set_header X-Forwarded-Proto $scheme;
}
}
Создадим службу в systemd
Description=Keenetic Wi-Fi Statistics
After=network.target mariadb.service
Requires=mariadb.service[Service]
Type=simple
User=root
WorkingDirectory=/opt/uat/stat
ExecStart=/opt/uat/venv/bin/python /opt/uat/stat/app_stat.py
Restart=always
RestartSec=3[Install]
WantedBy=multi-user.target
Включить автозапуск
systemctl enable keenetic-stat
systemctl start keenetic-stat
sudo systemctl status keenetic-stat
Смотреть журнал приложения
Осталось сделать возможность задавать имена точкам доступа
Создаём таблицу AP в MariaDB:
sudo mariadb radius
Проверка:
И:
GRANT SELECT, INSERT, UPDATE, DELETE ON radius.ap TO 'radius_stat'@'localhost';
FLUSH PRIVILEGES;
Пока таблица будет пустая.
Добавим AP из уже имеющихся radacct
INSERT INTO ap (mac, name, ip)
SELECT
calledstationid,
'',
MAX(nasipaddress)
FROM radacct
WHERE calledstationid <> ''
GROUP BY calledstationid
ON DUPLICATE KEY UPDATE
ip = VALUES(ip);
проверка
Теперь добавим веб-управление названиями AP в существующий app_stat.py (в моем случае app_stst.py уже был приведен к этому виду в предидущем пункте)
После функции db_connect() добавить:
@app.route("/stats/ap", methods=["GET", "POST"])
def ap_list():
if request.method == "POST":
ap_id = request.form.get("id", "").strip()
name = request.form.get("name", "").strip()
ip = request.form.get("ip", "").strip()
if ap_id:
with db_connect() as conn:
with conn.cursor() as cursor:
cursor.execute(
"""
UPDATE ap
SET name = %s, ip = %s
WHERE id = %s
""",
(name, ip or None, ap_id)
)
conn.commit()
with db_connect() as conn:
with conn.cursor() as cursor:
cursor.execute("""
SELECT id, mac, name, ip
FROM ap
ORDER BY mac
""")
aps = cursor.fetchall()
return render_template("ap.html", aps=aps)
Создаём страницу
nano /opt/uat/stat/ap.html
<html lang="ru">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<title>Точки доступа</title><style>
body {
font-family: Arial, sans-serif;
margin: 30px;
background: #f5f5f5;
}
h1 {
margin-bottom: 20px;
}
a {
color: #0066cc;
text-decoration: none;
}
table {
border-collapse: collapse;
width: 100%;
max-width: 900px;
background: white;
}
th, td {
border: 1px solid #ddd;
padding: 10px;
text-align: left;
}
th {
background: #eee;
}
input {
padding: 7px;
width: 90%;
box-sizing: border-box;
}
button {
padding: 7px 15px;
cursor: pointer;
}
.top {
margin-bottom: 20px;
}
</style>
</head>
<body>
<h1>Точки доступа</h1>
<div class="top">
<a href="/stats">← Статистика</a>
</div>
<table>
<thead>
<tr>
<th>MAC</th>
<th>Название</th>
<th>IP</th>
<th>Действие</th>
</tr>
</thead>
<tbody>
{% for ap in aps %}
<tr>
<form method="post">
<td>
{{ ap.mac }}
<input type="hidden" name="id" value="{{ ap.id }}">
</td>
<td>
<input
type="text"
name="name"
value="{{ ap.name }}"
placeholder="Название AP">
</td>
<td>
<input
type="text"
name="ip"
value="{{ ap.ip or '' }}"
placeholder="IP">
</td>
<td>
<button type="submit">Сохранить</button>
</td>
</form>
</tr>
{% endfor %}
</tbody>
</table>
</body>
</html>