Archived
218 lines
8.0 KiB
Python
218 lines
8.0 KiB
Python
from __future__ import annotations
|
|
|
|
from collections.abc import Mapping
|
|
from dataclasses import dataclass
|
|
|
|
from fastapi import APIRouter, Request, Response
|
|
from fastapi.responses import HTMLResponse, RedirectResponse
|
|
from starlette.concurrency import run_in_threadpool
|
|
|
|
from ..dependencies import ServiceDep, SettingsDep, StoreDep
|
|
from ..event_statuses import DASHBOARD_PASSWORD_CHANGED, DASHBOARD_PASSWORD_ENABLED, DASHBOARD_PASSWORD_REMOVED
|
|
from ..security import client_ip, dashboard_csrf_token, require_dashboard_csrf
|
|
from ..store import DdnsStore
|
|
from ..web.forms import first_form_value, request_form
|
|
from ..web.templates_management import ManagementPageContext, _render_management_unlock_page
|
|
from .management_shared import (
|
|
ManagementAccount,
|
|
delete_unlock_cookie,
|
|
get_active_management_account_async,
|
|
has_management_access,
|
|
management_error_response_async,
|
|
render_management_page_async,
|
|
set_unlock_cookie,
|
|
unlock_cookie_seconds,
|
|
)
|
|
from .path_params import ManagementSlugPath
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class _DashboardPasswordErrorContext:
|
|
management_slug: str
|
|
account: ManagementAccount
|
|
form: dict[str, list[str]]
|
|
request_ip: str
|
|
current_password_error: str | None
|
|
|
|
|
|
@router.post("/m/{management_slug}/unlock", include_in_schema=False)
|
|
async def unlock_dashboard(
|
|
management_slug: ManagementSlugPath, request: Request, settings: SettingsDep, store: StoreDep
|
|
) -> Response:
|
|
account = await get_active_management_account_async(store, management_slug)
|
|
if not account.get("dashboard_password_hash"):
|
|
return RedirectResponse(f"/m/{management_slug}", status_code=303)
|
|
form = await request_form(request, settings)
|
|
require_dashboard_csrf(management_slug, account, first_form_value(form, "csrf"))
|
|
password = first_form_value(form, "dashboard_password")
|
|
hostname = str(account["hostname"])
|
|
is_valid = await run_in_threadpool(store.verify_dashboard_password, hostname, password)
|
|
if not is_valid:
|
|
return HTMLResponse(
|
|
_render_management_unlock_page(
|
|
management_slug, account, dashboard_csrf_token(management_slug, account), "Incorrect password."
|
|
),
|
|
status_code=401,
|
|
)
|
|
response = RedirectResponse(f"/m/{management_slug}", status_code=303)
|
|
set_unlock_cookie(
|
|
response,
|
|
settings,
|
|
management_slug,
|
|
account,
|
|
unlock_cookie_seconds(settings, first_form_value(form, "stay_unlocked")),
|
|
)
|
|
return response
|
|
|
|
|
|
@router.post("/m/{management_slug}/password", include_in_schema=False)
|
|
async def set_dashboard_password(
|
|
management_slug: ManagementSlugPath,
|
|
request: Request,
|
|
settings: SettingsDep,
|
|
store: StoreDep,
|
|
service: ServiceDep,
|
|
) -> Response:
|
|
account = await get_active_management_account_async(store, management_slug)
|
|
if not has_management_access(request, settings, management_slug, account):
|
|
return HTMLResponse(
|
|
_render_management_unlock_page(management_slug, account, dashboard_csrf_token(management_slug, account)),
|
|
status_code=401,
|
|
)
|
|
form = await request_form(request, settings)
|
|
require_dashboard_csrf(management_slug, account, first_form_value(form, "csrf"))
|
|
hostname = str(account["hostname"])
|
|
current_password_error = await _verify_current_dashboard_password(store, hostname, account, form)
|
|
request_ip = client_ip(request, settings)
|
|
error_response = await _dashboard_password_error_response(
|
|
service,
|
|
_DashboardPasswordErrorContext(
|
|
management_slug=management_slug,
|
|
account=account,
|
|
form=form,
|
|
request_ip=request_ip,
|
|
current_password_error=current_password_error,
|
|
),
|
|
)
|
|
if error_response:
|
|
return error_response
|
|
password = first_form_value(form, "dashboard_password")
|
|
had_password = bool(account.get("dashboard_password_hash"))
|
|
await run_in_threadpool(store.set_dashboard_password, hostname, password)
|
|
account = await get_active_management_account_async(store, management_slug)
|
|
await run_in_threadpool(
|
|
store.log_update_event,
|
|
hostname,
|
|
None,
|
|
None,
|
|
DASHBOARD_PASSWORD_CHANGED if had_password else DASHBOARD_PASSWORD_ENABLED,
|
|
"dashboard password changed" if had_password else "dashboard password enabled",
|
|
client_ip(request, settings),
|
|
)
|
|
response = RedirectResponse(f"/m/{management_slug}", status_code=303)
|
|
set_unlock_cookie(response, settings, management_slug, account, settings.dashboard_unlock_seconds)
|
|
return response
|
|
|
|
|
|
def _dashboard_password_error(password: str, confirm: str) -> str | None:
|
|
if len(password) < 8:
|
|
return "Password must be at least 8 characters."
|
|
if password != confirm:
|
|
return "The passwords do not match."
|
|
return None
|
|
|
|
|
|
async def _dashboard_password_error_response(
|
|
service: ServiceDep,
|
|
context: _DashboardPasswordErrorContext,
|
|
) -> HTMLResponse | None:
|
|
if context.current_password_error:
|
|
return await management_error_response_async(
|
|
service,
|
|
context.management_slug,
|
|
context.account,
|
|
context.current_password_error,
|
|
context.request_ip,
|
|
)
|
|
password_error = _dashboard_password_error(
|
|
first_form_value(context.form, "dashboard_password"),
|
|
first_form_value(context.form, "dashboard_password_confirm"),
|
|
)
|
|
if password_error is None:
|
|
return None
|
|
page = await render_management_page_async(
|
|
service,
|
|
ManagementPageContext(
|
|
management_slug=context.management_slug,
|
|
account=context.account,
|
|
csrf_token=dashboard_csrf_token(context.management_slug, context.account),
|
|
message=password_error,
|
|
request_ip=context.request_ip,
|
|
),
|
|
)
|
|
return HTMLResponse(page, status_code=400)
|
|
|
|
|
|
async def _verify_current_dashboard_password(
|
|
store: DdnsStore,
|
|
hostname: str,
|
|
account: Mapping[str, object],
|
|
form: dict[str, list[str]],
|
|
) -> str | None:
|
|
if not account.get("dashboard_password_hash"):
|
|
return None
|
|
is_current = await run_in_threadpool(
|
|
store.verify_dashboard_password,
|
|
hostname,
|
|
first_form_value(form, "current_dashboard_password"),
|
|
)
|
|
return None if is_current else "Current password is incorrect."
|
|
|
|
|
|
@router.post("/m/{management_slug}/password/remove", include_in_schema=False)
|
|
async def remove_dashboard_password(
|
|
management_slug: ManagementSlugPath,
|
|
request: Request,
|
|
settings: SettingsDep,
|
|
store: StoreDep,
|
|
service: ServiceDep,
|
|
) -> Response:
|
|
account = await get_active_management_account_async(store, management_slug)
|
|
if not has_management_access(request, settings, management_slug, account):
|
|
return HTMLResponse(
|
|
_render_management_unlock_page(management_slug, account, dashboard_csrf_token(management_slug, account)),
|
|
status_code=401,
|
|
)
|
|
form = await request_form(request, settings)
|
|
require_dashboard_csrf(management_slug, account, first_form_value(form, "csrf"))
|
|
hostname = str(account["hostname"])
|
|
if account.get("dashboard_password_hash"):
|
|
is_current = await run_in_threadpool(
|
|
store.verify_dashboard_password,
|
|
hostname,
|
|
first_form_value(form, "current_dashboard_password"),
|
|
)
|
|
if not is_current:
|
|
return await management_error_response_async(
|
|
service,
|
|
management_slug,
|
|
account,
|
|
"Current password is incorrect.",
|
|
client_ip(request, settings),
|
|
)
|
|
await run_in_threadpool(store.clear_dashboard_password, hostname)
|
|
await run_in_threadpool(
|
|
store.log_update_event,
|
|
hostname,
|
|
None,
|
|
None,
|
|
DASHBOARD_PASSWORD_REMOVED,
|
|
"dashboard password removed",
|
|
client_ip(request, settings),
|
|
)
|
|
response = RedirectResponse(f"/m/{management_slug}", status_code=303)
|
|
delete_unlock_cookie(response, management_slug)
|
|
return response
|