Update project configuration and templates
This commit is contained in:
AVS
@@ -18,6 +18,7 @@ from sqlalchemy.orm import Session
|
||||
from starlette.middleware.sessions import SessionMiddleware
|
||||
|
||||
from config import (
|
||||
ADMIN_PANEL_ENABLED,
|
||||
ADMIN_SESSION_SECRET,
|
||||
ALLOWED_ADMIN_IPS,
|
||||
DATABASE_URL,
|
||||
@@ -206,6 +207,13 @@ async def lifespan(app: FastAPI):
|
||||
logger.info(f"Created default admin user: {admin.username}")
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
if ADMIN_PANEL_ENABLED:
|
||||
register_admin_routes(app)
|
||||
logger.warning("Admin panel ENABLED - configure apps then disable!")
|
||||
else:
|
||||
logger.info("Admin panel DISABLED (ADMIN_PANEL_ENABLED=false)")
|
||||
|
||||
logger.info("TNProxy started")
|
||||
yield
|
||||
|
||||
@@ -226,6 +234,286 @@ def get_application_by_token(token: str, db: Session) -> Optional[Application]:
|
||||
return db.query(Application).filter(Application.token == token).first()
|
||||
|
||||
|
||||
def register_admin_routes(app: FastAPI):
|
||||
@app.get("/logtn", response_class=HTMLResponse)
|
||||
async def login_page(request: Request, error: str = None):
|
||||
html = render_template("logtn.html", {"request": request, "error": error})
|
||||
return HTMLResponse(html)
|
||||
|
||||
@app.post("/logtn")
|
||||
async def login(request: Request, login_data: LoginRequest):
|
||||
db = SessionLocal()
|
||||
try:
|
||||
admin = (
|
||||
db.query(AdminUser)
|
||||
.filter(AdminUser.username == login_data.username)
|
||||
.first()
|
||||
)
|
||||
|
||||
client_ip = request.client.host if request.client else "unknown"
|
||||
|
||||
if admin and admin.check_password(login_data.password):
|
||||
admin.last_login = datetime.utcnow()
|
||||
db.commit()
|
||||
request.session["admin_user"] = admin.username
|
||||
await notify_login_success(login_data.username, client_ip)
|
||||
logger.info(f"Admin login: {login_data.username} from {client_ip}")
|
||||
return {"success": True, "redirect": "/adtn"}
|
||||
else:
|
||||
await notify_login_failure(login_data.username, client_ip)
|
||||
logger.warning(
|
||||
f"Failed login attempt: {login_data.username} from {client_ip}"
|
||||
)
|
||||
return {"success": False, "error": "Invalid credentials"}
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
@app.post("/logout")
|
||||
async def logout(request: Request):
|
||||
request.session.clear()
|
||||
return {"success": True, "redirect": "/logtn"}
|
||||
|
||||
@app.get("/adtn", response_class=HTMLResponse)
|
||||
async def admin_dashboard(request: Request):
|
||||
try:
|
||||
require_auth(request)
|
||||
except HTTPException:
|
||||
return Response(headers={"Location": "/logtn"}, status_code=302)
|
||||
|
||||
db = SessionLocal()
|
||||
try:
|
||||
apps_list = [a.to_dict() for a in db.query(Application).all()]
|
||||
log_settings = get_log_settings(db)
|
||||
finally:
|
||||
db.close()
|
||||
html = render_template(
|
||||
"index.html",
|
||||
{"request": request, "apps": apps_list, "settings": log_settings},
|
||||
)
|
||||
return HTMLResponse(html)
|
||||
|
||||
@app.get("/adtn/password", response_class=HTMLResponse)
|
||||
async def password_page(request: Request):
|
||||
try:
|
||||
require_auth(request)
|
||||
except HTTPException:
|
||||
return Response(headers={"Location": "/logtn"}, status_code=302)
|
||||
|
||||
html = render_template("password.html", {"request": request})
|
||||
return HTMLResponse(html)
|
||||
|
||||
@app.post("/adtn/password")
|
||||
async def change_password(request: Request, password_data: PasswordChange):
|
||||
try:
|
||||
require_auth(request)
|
||||
except HTTPException:
|
||||
return {"success": False, "error": "Not authenticated"}
|
||||
|
||||
db = SessionLocal()
|
||||
try:
|
||||
admin = db.query(AdminUser).first()
|
||||
if not admin.check_password(password_data.old_password):
|
||||
return {"success": False, "error": "Current password is incorrect"}
|
||||
|
||||
admin.password_hash = AdminUser.hash_password(password_data.new_password)
|
||||
db.commit()
|
||||
logger.info(f"Password changed for user: {admin.username}")
|
||||
return {"success": True}
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
@app.get("/adtn/api", response_class=HTMLResponse)
|
||||
async def admin_api_docs(request: Request):
|
||||
try:
|
||||
require_auth(request)
|
||||
except HTTPException:
|
||||
return Response(headers={"Location": "/logtn"}, status_code=302)
|
||||
|
||||
html = render_template("api.html", {"request": request})
|
||||
return HTMLResponse(html)
|
||||
|
||||
@app.post("/adtn/applications")
|
||||
async def create_application(request: Request, app_data: ApplicationCreate):
|
||||
try:
|
||||
require_auth(request)
|
||||
except HTTPException:
|
||||
raise HTTPException(status_code=401, detail="Not authenticated")
|
||||
|
||||
db = SessionLocal()
|
||||
try:
|
||||
new_app = Application(
|
||||
name=app_data.name,
|
||||
token=Application.generate_token(),
|
||||
telegram_bot_token=app_data.telegram_bot_token,
|
||||
telegram_chat_id=app_data.telegram_chat_id,
|
||||
)
|
||||
db.add(new_app)
|
||||
db.commit()
|
||||
db.refresh(new_app)
|
||||
logger.info(f"Created application: {new_app.name}")
|
||||
return {"id": new_app.id, "token": new_app.token}
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
@app.delete("/adtn/applications/{app_id}")
|
||||
async def delete_application(request: Request, app_id: int):
|
||||
try:
|
||||
require_auth(request)
|
||||
except HTTPException:
|
||||
raise HTTPException(status_code=401, detail="Not authenticated")
|
||||
|
||||
db = SessionLocal()
|
||||
try:
|
||||
app = db.query(Application).filter(Application.id == app_id).first()
|
||||
if not app:
|
||||
raise HTTPException(status_code=404, detail="Application not found")
|
||||
db.delete(app)
|
||||
db.commit()
|
||||
logger.info(f"Deleted application: {app.name}")
|
||||
return {"success": True}
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
@app.post("/adtn/applications/{app_id}/toggle")
|
||||
async def toggle_application(request: Request, app_id: int):
|
||||
try:
|
||||
require_auth(request)
|
||||
except HTTPException:
|
||||
raise HTTPException(status_code=401, detail="Not authenticated")
|
||||
|
||||
db = SessionLocal()
|
||||
try:
|
||||
app = db.query(Application).filter(Application.id == app_id).first()
|
||||
if not app:
|
||||
raise HTTPException(status_code=404, detail="Application not found")
|
||||
app.is_active = not app.is_active
|
||||
db.commit()
|
||||
logger.info(f"Toggled application {app.name}: is_active={app.is_active}")
|
||||
return {"is_active": app.is_active}
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
@app.get("/adtn/logs", response_class=HTMLResponse)
|
||||
async def admin_logs(request: Request):
|
||||
try:
|
||||
require_auth(request)
|
||||
except HTTPException:
|
||||
return Response(headers={"Location": "/logtn"}, status_code=302)
|
||||
|
||||
db = SessionLocal()
|
||||
try:
|
||||
log_settings = get_log_settings(db)
|
||||
finally:
|
||||
db.close()
|
||||
html = render_template(
|
||||
"logs.html",
|
||||
{"request": request, "settings": log_settings},
|
||||
)
|
||||
return HTMLResponse(html)
|
||||
|
||||
@app.post("/adtn/logs")
|
||||
async def update_log_settings(request: Request, settings: LogSettingsUpdate):
|
||||
try:
|
||||
require_auth(request)
|
||||
except HTTPException:
|
||||
raise HTTPException(status_code=401, detail="Not authenticated")
|
||||
|
||||
db = SessionLocal()
|
||||
try:
|
||||
log_settings = get_log_settings(db)
|
||||
log_settings.max_bytes = settings.max_bytes
|
||||
log_settings.backup_count = settings.backup_count
|
||||
log_settings.log_level = settings.log_level
|
||||
db.commit()
|
||||
|
||||
update_logging_config(log_settings)
|
||||
logger.info(
|
||||
f"Updated log settings: level={log_settings.log_level}, max_bytes={log_settings.max_bytes}, backup_count={log_settings.backup_count}"
|
||||
)
|
||||
return {"success": True}
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
@app.get("/adtn/export", response_class=HTMLResponse)
|
||||
async def export_settings(request: Request):
|
||||
try:
|
||||
require_auth(request)
|
||||
except HTTPException:
|
||||
return Response(headers={"Location": "/logtn"}, status_code=302)
|
||||
|
||||
db = SessionLocal()
|
||||
try:
|
||||
apps_list = db.query(Application).all()
|
||||
log_settings = get_log_settings(db)
|
||||
export_data = {
|
||||
"applications": [a.to_dict() for a in apps_list],
|
||||
"log_settings": {
|
||||
"log_level": log_settings.log_level,
|
||||
"max_bytes": log_settings.max_bytes,
|
||||
"backup_count": log_settings.backup_count,
|
||||
},
|
||||
}
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
import json
|
||||
from fastapi.responses import Response
|
||||
|
||||
return Response(
|
||||
content=json.dumps(export_data, indent=2),
|
||||
media_type="application/json",
|
||||
headers={"Content-Disposition": "attachment; filename=tnproxy_backup.json"},
|
||||
)
|
||||
|
||||
@app.post("/adtn/import")
|
||||
async def import_settings(request: Request):
|
||||
try:
|
||||
require_auth(request)
|
||||
except HTTPException:
|
||||
raise HTTPException(status_code=401, detail="Not authenticated")
|
||||
|
||||
import json
|
||||
|
||||
try:
|
||||
data = await request.json()
|
||||
except Exception:
|
||||
return {"success": False, "error": "Invalid JSON"}
|
||||
|
||||
db = SessionLocal()
|
||||
try:
|
||||
if "applications" in data:
|
||||
for app_data in data["applications"]:
|
||||
existing = (
|
||||
db.query(Application)
|
||||
.filter(Application.token == app_data.get("token"))
|
||||
.first()
|
||||
)
|
||||
if not existing:
|
||||
new_app = Application(
|
||||
name=app_data["name"],
|
||||
token=app_data["token"],
|
||||
telegram_bot_token=app_data["telegram_bot_token"],
|
||||
telegram_chat_id=app_data["telegram_chat_id"],
|
||||
is_active=app_data.get("is_active", True),
|
||||
)
|
||||
db.add(new_app)
|
||||
|
||||
if "log_settings" in data:
|
||||
log_settings = get_log_settings(db)
|
||||
log_settings.log_level = data["log_settings"].get("log_level", "INFO")
|
||||
log_settings.max_bytes = data["log_settings"].get("max_bytes", 10485760)
|
||||
log_settings.backup_count = data["log_settings"].get("backup_count", 5)
|
||||
|
||||
db.commit()
|
||||
logger.info("Settings imported successfully")
|
||||
return {"success": True}
|
||||
except Exception as e:
|
||||
logger.error(f"Import error: {e}")
|
||||
return {"success": False, "error": str(e)}
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
|
||||
@app.get("/health")
|
||||
async def health():
|
||||
return {"status": "ok"}
|
||||
@@ -282,297 +570,5 @@ async def send_message(
|
||||
db.close()
|
||||
|
||||
|
||||
@app.get("/logtn", response_class=HTMLResponse)
|
||||
async def login_page(request: Request, error: str = None):
|
||||
html = render_template("logtn.html", {"request": request, "error": error})
|
||||
return HTMLResponse(html)
|
||||
|
||||
|
||||
@app.post("/logtn")
|
||||
async def login(request: Request, login_data: LoginRequest):
|
||||
db = SessionLocal()
|
||||
try:
|
||||
admin = (
|
||||
db.query(AdminUser)
|
||||
.filter(AdminUser.username == login_data.username)
|
||||
.first()
|
||||
)
|
||||
|
||||
client_ip = request.client.host if request.client else "unknown"
|
||||
|
||||
if admin and admin.check_password(login_data.password):
|
||||
admin.last_login = datetime.utcnow()
|
||||
db.commit()
|
||||
request.session["admin_user"] = admin.username
|
||||
await notify_login_success(login_data.username, client_ip)
|
||||
logger.info(f"Admin login: {login_data.username} from {client_ip}")
|
||||
return {"success": True, "redirect": "/adtn"}
|
||||
else:
|
||||
await notify_login_failure(login_data.username, client_ip)
|
||||
logger.warning(
|
||||
f"Failed login attempt: {login_data.username} from {client_ip}"
|
||||
)
|
||||
return {"success": False, "error": "Invalid credentials"}
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
|
||||
@app.post("/logout")
|
||||
async def logout(request: Request):
|
||||
request.session.clear()
|
||||
return {"success": True, "redirect": "/logtn"}
|
||||
|
||||
|
||||
@app.get("/adtn", response_class=HTMLResponse)
|
||||
async def admin_dashboard(request: Request):
|
||||
try:
|
||||
require_auth(request)
|
||||
except HTTPException:
|
||||
return Response(headers={"Location": "/logtn"}, status_code=302)
|
||||
|
||||
db = SessionLocal()
|
||||
try:
|
||||
apps = [app.to_dict() for app in db.query(Application).all()]
|
||||
log_settings = get_log_settings(db)
|
||||
finally:
|
||||
db.close()
|
||||
html = render_template(
|
||||
"index.html",
|
||||
{"request": request, "apps": apps, "settings": log_settings},
|
||||
)
|
||||
return HTMLResponse(html)
|
||||
|
||||
|
||||
@app.get("/adtn/password", response_class=HTMLResponse)
|
||||
async def password_page(request: Request):
|
||||
try:
|
||||
require_auth(request)
|
||||
except HTTPException:
|
||||
return Response(headers={"Location": "/logtn"}, status_code=302)
|
||||
|
||||
html = render_template("password.html", {"request": request})
|
||||
return HTMLResponse(html)
|
||||
|
||||
|
||||
@app.post("/adtn/password")
|
||||
async def change_password(request: Request, password_data: PasswordChange):
|
||||
try:
|
||||
require_auth(request)
|
||||
except HTTPException:
|
||||
return {"success": False, "error": "Not authenticated"}
|
||||
|
||||
db = SessionLocal()
|
||||
try:
|
||||
admin = db.query(AdminUser).first()
|
||||
if not admin.check_password(password_data.old_password):
|
||||
return {"success": False, "error": "Current password is incorrect"}
|
||||
|
||||
admin.password_hash = AdminUser.hash_password(password_data.new_password)
|
||||
db.commit()
|
||||
logger.info(f"Password changed for user: {admin.username}")
|
||||
return {"success": True}
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
|
||||
@app.get("/adtn/api", response_class=HTMLResponse)
|
||||
async def admin_api_docs(request: Request):
|
||||
try:
|
||||
require_auth(request)
|
||||
except HTTPException:
|
||||
return Response(headers={"Location": "/logtn"}, status_code=302)
|
||||
|
||||
html = render_template("api.html", {"request": request})
|
||||
return HTMLResponse(html)
|
||||
|
||||
|
||||
@app.post("/adtn/applications")
|
||||
async def create_application(request: Request, app_data: ApplicationCreate):
|
||||
try:
|
||||
require_auth(request)
|
||||
except HTTPException:
|
||||
raise HTTPException(status_code=401, detail="Not authenticated")
|
||||
|
||||
db = SessionLocal()
|
||||
try:
|
||||
app = Application(
|
||||
name=app_data.name,
|
||||
token=Application.generate_token(),
|
||||
telegram_bot_token=app_data.telegram_bot_token,
|
||||
telegram_chat_id=app_data.telegram_chat_id,
|
||||
)
|
||||
db.add(app)
|
||||
db.commit()
|
||||
db.refresh(app)
|
||||
logger.info(f"Created application: {app.name}")
|
||||
return {"id": app.id, "token": app.token}
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
|
||||
@app.delete("/adtn/applications/{app_id}")
|
||||
async def delete_application(request: Request, app_id: int):
|
||||
try:
|
||||
require_auth(request)
|
||||
except HTTPException:
|
||||
raise HTTPException(status_code=401, detail="Not authenticated")
|
||||
|
||||
db = SessionLocal()
|
||||
try:
|
||||
app = db.query(Application).filter(Application.id == app_id).first()
|
||||
if not app:
|
||||
raise HTTPException(status_code=404, detail="Application not found")
|
||||
db.delete(app)
|
||||
db.commit()
|
||||
logger.info(f"Deleted application: {app.name}")
|
||||
return {"success": True}
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
|
||||
@app.post("/adtn/applications/{app_id}/toggle")
|
||||
async def toggle_application(request: Request, app_id: int):
|
||||
try:
|
||||
require_auth(request)
|
||||
except HTTPException:
|
||||
raise HTTPException(status_code=401, detail="Not authenticated")
|
||||
|
||||
db = SessionLocal()
|
||||
try:
|
||||
app = db.query(Application).filter(Application.id == app_id).first()
|
||||
if not app:
|
||||
raise HTTPException(status_code=404, detail="Application not found")
|
||||
app.is_active = not app.is_active
|
||||
db.commit()
|
||||
logger.info(f"Toggled application {app.name}: is_active={app.is_active}")
|
||||
return {"is_active": app.is_active}
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
|
||||
@app.get("/adtn/logs", response_class=HTMLResponse)
|
||||
async def admin_logs(request: Request):
|
||||
try:
|
||||
require_auth(request)
|
||||
except HTTPException:
|
||||
return Response(headers={"Location": "/logtn"}, status_code=302)
|
||||
|
||||
db = SessionLocal()
|
||||
try:
|
||||
log_settings = get_log_settings(db)
|
||||
finally:
|
||||
db.close()
|
||||
html = render_template(
|
||||
"logs.html",
|
||||
{"request": request, "settings": log_settings},
|
||||
)
|
||||
return HTMLResponse(html)
|
||||
|
||||
|
||||
@app.post("/adtn/logs")
|
||||
async def update_log_settings(request: Request, settings: LogSettingsUpdate):
|
||||
try:
|
||||
require_auth(request)
|
||||
except HTTPException:
|
||||
raise HTTPException(status_code=401, detail="Not authenticated")
|
||||
|
||||
db = SessionLocal()
|
||||
try:
|
||||
log_settings = get_log_settings(db)
|
||||
log_settings.max_bytes = settings.max_bytes
|
||||
log_settings.backup_count = settings.backup_count
|
||||
log_settings.log_level = settings.log_level
|
||||
db.commit()
|
||||
|
||||
update_logging_config(log_settings)
|
||||
logger.info(
|
||||
f"Updated log settings: level={log_settings.log_level}, max_bytes={log_settings.max_bytes}, backup_count={log_settings.backup_count}"
|
||||
)
|
||||
return {"success": True}
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
|
||||
@app.get("/adtn/export", response_class=HTMLResponse)
|
||||
async def export_settings(request: Request):
|
||||
try:
|
||||
require_auth(request)
|
||||
except HTTPException:
|
||||
return Response(headers={"Location": "/logtn"}, status_code=302)
|
||||
|
||||
db = SessionLocal()
|
||||
try:
|
||||
apps = db.query(Application).all()
|
||||
log_settings = get_log_settings(db)
|
||||
export_data = {
|
||||
"applications": [app.to_dict() for app in apps],
|
||||
"log_settings": {
|
||||
"log_level": log_settings.log_level,
|
||||
"max_bytes": log_settings.max_bytes,
|
||||
"backup_count": log_settings.backup_count,
|
||||
},
|
||||
}
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
import json
|
||||
from fastapi.responses import Response
|
||||
|
||||
return Response(
|
||||
content=json.dumps(export_data, indent=2),
|
||||
media_type="application/json",
|
||||
headers={"Content-Disposition": "attachment; filename=tnproxy_backup.json"},
|
||||
)
|
||||
|
||||
|
||||
@app.post("/adtn/import")
|
||||
async def import_settings(request: Request):
|
||||
try:
|
||||
require_auth(request)
|
||||
except HTTPException:
|
||||
raise HTTPException(status_code=401, detail="Not authenticated")
|
||||
|
||||
import json
|
||||
|
||||
try:
|
||||
data = await request.json()
|
||||
except Exception:
|
||||
return {"success": False, "error": "Invalid JSON"}
|
||||
|
||||
db = SessionLocal()
|
||||
try:
|
||||
if "applications" in data:
|
||||
for app_data in data["applications"]:
|
||||
existing = (
|
||||
db.query(Application)
|
||||
.filter(Application.token == app_data.get("token"))
|
||||
.first()
|
||||
)
|
||||
if not existing:
|
||||
app = Application(
|
||||
name=app_data["name"],
|
||||
token=app_data["token"],
|
||||
telegram_bot_token=app_data["telegram_bot_token"],
|
||||
telegram_chat_id=app_data["telegram_chat_id"],
|
||||
is_active=app_data.get("is_active", True),
|
||||
)
|
||||
db.add(app)
|
||||
|
||||
if "log_settings" in data:
|
||||
log_settings = get_log_settings(db)
|
||||
log_settings.log_level = data["log_settings"].get("log_level", "INFO")
|
||||
log_settings.max_bytes = data["log_settings"].get("max_bytes", 10485760)
|
||||
log_settings.backup_count = data["log_settings"].get("backup_count", 5)
|
||||
|
||||
db.commit()
|
||||
logger.info("Settings imported successfully")
|
||||
return {"success": True}
|
||||
except Exception as e:
|
||||
logger.error(f"Import error: {e}")
|
||||
return {"success": False, "error": str(e)}
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
uvicorn.run(app, host=SERVER_HOST, port=SERVER_PORT)
|
||||
|
||||
Reference in New Issue
Block a user