a.gruzdev/Site_IKP_zaglushka
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: аутентификация через Authentik OIDC вместо пароля

Browse Source
This commit is contained in:
AGorokhov
2026-06-21 20:57:00 +03:00
parent b0c9cb91b4
commit 12bee9dce4
2 changed files with 106 additions and 110 deletions
Download Patch File Download Diff File Expand all files Collapse all files
config.py
+8 -2
View File
@@ -48,8 +48,14 @@ SMTP_PASSWORD = os.getenv("SMTP_PASSWORD", "")
SMTP_FROM = os.getenv("SMTP_FROM", "")
SMTP_TO = os.getenv("SMTP_TO", "")
# Admin Dashboard Password
ADMIN_PASSWORD = os.getenv("ADMIN_PASSWORD", "admin123")
# Admin Dashboard Password (устарело — используется OIDC через Authentik)
ADMIN_PASSWORD = os.getenv("ADMIN_PASSWORD", "")
# Authentik OIDC Configuration
AUTHENTIK_BASE_URL = os.getenv("AUTHENTIK_BASE_URL", "") # например https://auth.infocyber.pro
AUTHENTIK_CLIENT_ID = os.getenv("AUTHENTIK_CLIENT_ID", "")
AUTHENTIK_CLIENT_SECRET = os.getenv("AUTHENTIK_CLIENT_SECRET", "")
AUTHENTIK_REDIRECT_URI = os.getenv("AUTHENTIK_REDIRECT_URI", "") # например https://infocyber.pro/admin/callback
# Supplier / Agent Configuration for 54-FZ receipts (Agent Scheme)
SUPPLIER_NAME = os.getenv("SUPPLIER_NAME", "ИП Груздев Артём Сергеевич")
server.py
+91 -101
View File
@@ -691,47 +691,27 @@ def yookassa_webhook():
# ====================================================
# Admin Cabinet Views (ЛК Администратора)
# ====================================================
LOGIN_TEMPLATE = """
# Шаблон страницы ошибки доступа
ACCESS_DENIED_TEMPLATE = """
<!DOCTYPE html>
<html lang="ru">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>ИКП Панель | Авторизация</title>
<title>ИКП Админ | Ошибка авторизации</title>
<script src="https://cdn.tailwindcss.com"></script>
<link href="https://fonts.googleapis.com/css2?family=Inter:wght@400;500;600&family=Montserrat:wght@700;800&display=swap" rel="stylesheet">
<style>
body { font-family: 'Inter', sans-serif; background-color: #0a0a0f; }
.font-display { font-family: 'Montserrat', sans-serif; }
</style>
<style>body { font-family: 'Inter', sans-serif; background-color: #0a0a0f; }</style>
</head>
<body class="min-h-screen flex items-center justify-center relative overflow-hidden bg-[radial-gradient(circle_at_center,_rgba(124,58,237,0.1)_0%,_rgba(10,10,15,1)_100%)]">
<div class="absolute -top-40 -left-40 w-96 h-96 bg-purple-600/10 rounded-full blur-[100px] pointer-events-none"></div>
<div class="absolute -bottom-40 -right-40 w-96 h-96 bg-blue-600/10 rounded-full blur-[100px] pointer-events-none"></div>
<div class="w-full max-w-md p-8 bg-slate-900/60 backdrop-blur-xl border border-white/10 rounded-2xl shadow-2xl relative z-10">
<div class="text-center mb-8">
<h2 class="font-display font-black text-3xl text-white tracking-wider uppercase mb-2">ИКП Админ</h2>
<p class="text-xs font-mono text-purple-400 uppercase tracking-widest">Панель управления оплатами</p>
</div>
{% if error %}
<div class="mb-6 p-4 bg-red-950/50 border border-red-500/30 text-red-200 text-sm rounded-lg flex items-center gap-2">
<span>⚠️</span> {{ error }}
</div>
{% endif %}
<form action="{{ url_for('admin_login') }}" method="POST" class="space-y-6">
<div>
<label for="password" class="block text-xs font-bold text-blue-400 uppercase tracking-wider mb-2">Пароль администратора</label>
<input type="password" id="password" name="password" required placeholder="••••••••"
class="w-full px-4 py-3 bg-black/50 border border-white/10 rounded-xl text-white placeholder-slate-600 focus:outline-none focus:border-purple-500 focus:ring-1 focus:ring-purple-500 transition">
</div>
<button type="submit" class="w-full py-3 bg-purple-600 hover:bg-purple-500 text-white font-display font-black text-sm uppercase tracking-wider rounded-xl shadow-lg shadow-purple-600/20 hover:shadow-purple-600/40 transform hover:-translate-y-0.5 transition duration-200">
Войти в кабинет
</button>
</form>
<body class="min-h-screen flex items-center justify-center bg-[radial-gradient(circle_at_center,_rgba(124,58,237,0.1)_0%,_rgba(10,10,15,1)_100%)]">
<div class="w-full max-w-md p-8 bg-slate-900/60 backdrop-blur-xl border border-white/10 rounded-2xl shadow-2xl text-center">
<div class="text-5xl mb-4">🔒</div>
<h2 class="font-display font-black text-2xl text-white uppercase mb-2" style="font-family:'Montserrat',sans-serif">Ошибка доступа</h2>
<p class="text-slate-400 text-sm mb-6">{{ error }}</p>
<a href="{{ login_url }}" class="inline-block w-full py-3 bg-purple-600 hover:bg-purple-500 text-white font-bold text-sm uppercase tracking-wider rounded-xl transition">
Войти через Authentik
</a>
</div>
</body>
</html>
@@ -1227,98 +1207,108 @@ DASHBOARD_TEMPLATE = """
@app.route('/admin', methods=['GET'])
def admin_dashboard():
if not session.get('logged_in'):
return render_template_string(LOGIN_TEMPLATE)
# Генерируем state для защиты от CSRF
state = str(uuid.uuid4())
session['oauth_state'] = state
# Fetch registrations and services
base = config.AUTHENTIK_BASE_URL.rstrip('/')
params = {
'client_id': config.AUTHENTIK_CLIENT_ID,
'redirect_uri': config.AUTHENTIK_REDIRECT_URI,
'response_type': 'code',
'scope': 'openid email profile',
'state': state,
}
auth_url = f"{base}/application/o/authorize/?" + "&".join(f"{k}={v}" for k, v in params.items())
return redirect(auth_url)
# Загрузить данные
registrations = db.get_all_web_registrations()
services = db.get_all_services_list()
# Calculate stats
stats = {
"total": len(registrations),
"paid_count": sum(1 for r in registrations if r["payment_status"] == "paid"),
"pending_count": sum(1 for r in registrations if r["payment_status"] != "paid"),
"total_revenue": sum(r["amount"] for r in registrations if r["payment_status"] == "paid")
}
return render_template_string(DASHBOARD_TEMPLATE, registrations=registrations, stats=stats, services=services)
@app.route('/admin/services/add', methods=['POST'])
def admin_add_service():
if not session.get('logged_in'):
return jsonify({"status": "error", "message": "Не авторизован"}), 403
@app.route('/admin/callback', methods=['GET'])
def admin_callback():
"""OIDC callback от Authentik."""
error = request.args.get('error')
if error:
print(f"[OIDC] Ошибка авторизации: {error} — {request.args.get('error_description', '')}")
base = config.AUTHENTIK_BASE_URL.rstrip('/')
login_url = f"{base}/application/o/authorize/?client_id={config.AUTHENTIK_CLIENT_ID}&redirect_uri={config.AUTHENTIK_REDIRECT_URI}&response_type=code&scope=openid+email+profile"
return render_template_string(ACCESS_DENIED_TEMPLATE,
error=f"Ошибка от Authentik: {error}",
login_url=login_url), 403
# Проверка state против CSRF
state = request.args.get('state')
if not state or state != session.get('oauth_state'):
print("[OIDC] Неверный state — возможная CSRF-атака")
return "403 Invalid state", 403
session.pop('oauth_state', None)
code = request.args.get('code')
if not code:
return "403 No code returned", 403
# Обмен кода на токен
base = config.AUTHENTIK_BASE_URL.rstrip('/')
token_url = f"{base}/application/o/token/"
try:
data = request.json
if not data:
return jsonify({"status": "error", "message": "Отсутствуют данные запроса"}), 400
service_id = data.get("id", "").strip()
name = data.get("name", "").strip()
price = data.get("price", "").strip()
price_numeric = data.get("price_numeric")
description = data.get("description", "").strip()
if not service_id or not name or not price or price_numeric is None:
return jsonify({"status": "error", "message": "Заполните все обязательные поля"}), 400
try:
price_numeric = float(price_numeric)
except ValueError:
return jsonify({"status": "error", "message": "Числовая цена должна быть числом"}), 400
success = db.add_service(service_id, name, price, price_numeric, description)
if success:
return jsonify({"status": "success", "message": "Услуга успешно добавлена"})
else:
return jsonify({"status": "error", "message": "Услуга с таким ID уже существует"}), 400
token_resp = requests.post(token_url, data={
'grant_type': 'authorization_code',
'code': code,
'redirect_uri': config.AUTHENTIK_REDIRECT_URI,
'client_id': config.AUTHENTIK_CLIENT_ID,
'client_secret': config.AUTHENTIK_CLIENT_SECRET,
}, timeout=15)
token_data = token_resp.json()
except Exception as e:
return jsonify({"status": "error", "message": str(e)}), 500
print(f"[OIDC] Ошибка получения токена: {e}")
return "500 Token exchange failed", 500
@app.route('/admin/services/edit/<service_id>', methods=['POST'])
def admin_edit_service(service_id):
if not session.get('logged_in'):
return jsonify({"status": "error", "message": "Не авторизован"}), 403
if 'error' in token_data:
print(f"[OIDC] Ошибка токена: {token_data}")
return f"403 {token_data.get('error_description', token_data['error'])}", 403
access_token = token_data.get('access_token')
# Получаем данные пользователя
try:
data = request.json
if not data:
return jsonify({"status": "error", "message": "Отсутствуют данные запроса"}), 400
name = data.get("name", "").strip()
price = data.get("price", "").strip()
price_numeric = data.get("price_numeric")
description = data.get("description", "").strip()
is_active = data.get("is_active", 1)
if not name or not price or price_numeric is None:
return jsonify({"status": "error", "message": "Заполните все обязательные поля"}), 400
try:
price_numeric = float(price_numeric)
except ValueError:
return jsonify({"status": "error", "message": "Числовая цена должна быть числом"}), 400
success = db.update_service(service_id, name, price, price_numeric, description, int(is_active))
if success:
return jsonify({"status": "success", "message": "Услуга успешно сохранена"})
else:
return jsonify({"status": "error", "message": "Не удалось обновить услугу"}), 400
userinfo_resp = requests.get(f"{base}/application/o/userinfo/",
headers={'Authorization': f'Bearer {access_token}'},
timeout=10)
userinfo = userinfo_resp.json()
except Exception as e:
return jsonify({"status": "error", "message": str(e)}), 500
print(f"[OIDC] Ошибка получения userinfo: {e}")
return "500 Userinfo request failed", 500
@app.route('/admin/login', methods=['POST'])
def admin_login():
password = request.form.get("password")
if password == config.ADMIN_PASSWORD:
print(f"[OIDC] Успешный вход: {userinfo.get('email') or userinfo.get('preferred_username', '?')}")
session['logged_in'] = True
session['user_email'] = userinfo.get('email', '')
session['user_name'] = userinfo.get('name') or userinfo.get('preferred_username', '')
return redirect(url_for('admin_dashboard'))
else:
return render_template_string(LOGIN_TEMPLATE, error="Неверный пароль администратора")
@app.route('/admin/login', methods=['POST', 'GET'])
def admin_login():
"""Legacy-маршрут — перенаправляет на OIDC."""
return redirect(url_for('admin_dashboard'))
@app.route('/admin/logout', methods=['GET'])
def admin_logout():
session.pop('logged_in', None)
session.clear()
# Выход также из Authentik
base = config.AUTHENTIK_BASE_URL.rstrip('/')
if base:
return redirect(f"{base}/application/o/ikp-admin/end-session/")
return redirect(url_for('admin_dashboard'))
@app.route('/admin/check-payment/<reg_id>', methods=['POST'])