A R R R Associates
796 lines
28 KiB
Python
796 lines
28 KiB
Python
from __future__ import annotations
|
|
|
|
from fastapi import APIRouter, Form, Request
|
|
from fastapi.responses import RedirectResponse
|
|
from sqlalchemy import select
|
|
|
|
from app.core.db.common import CommonSessionLocal
|
|
from app.core.security.csrf import get_or_create_csrf_token, validate_csrf
|
|
from app.core.security.session_auth import get_current_user
|
|
from app.core.templating import templates
|
|
from app.modules.core.rbac.deps import get_user_permissions, get_user_roles
|
|
from app.modules.domain_management.models import DomainMapping
|
|
from app.modules.domain_management.coolify_client import get_coolify_application_domains, sync_domain_to_coolify
|
|
from app.modules.domain_management.services import (
|
|
DomainMappingPayload,
|
|
create_domain_mapping,
|
|
create_or_get_tenant_subdomain_mapping,
|
|
create_or_get_firm_custom_domain_mapping,
|
|
create_or_get_consultant_domain_mapping,
|
|
list_consultant_domain_candidates,
|
|
list_domain_mappings,
|
|
list_domains_requiring_verification,
|
|
verify_all_pending_domains,
|
|
verify_domain_dns_txt,
|
|
build_ssl_proxy_snippets,
|
|
check_domain_ssl_certificate,
|
|
list_ssl_domains,
|
|
mark_ssl_managed,
|
|
list_tenant_subdomain_candidates,
|
|
mark_verified,
|
|
reference_data,
|
|
regenerate_verification_token,
|
|
update_domain_mapping,
|
|
validate_domain_payload,
|
|
)
|
|
|
|
router = APIRouter(prefix="/domains", tags=["domain-management-ui"])
|
|
|
|
|
|
def _is_domain_admin(db, user) -> bool:
|
|
roles = set(get_user_roles(db, user.id))
|
|
return "System Admin" in roles
|
|
|
|
|
|
def _base_ctx(request: Request, db, user, **ctx):
|
|
base = {
|
|
"request": request,
|
|
"current_user": user,
|
|
"current_user_roles": get_user_roles(db, user.id),
|
|
"current_user_permissions": get_user_permissions(db, user.id),
|
|
"csrf_token": get_or_create_csrf_token(request),
|
|
}
|
|
base.update(ctx)
|
|
return base
|
|
|
|
|
|
def _require_user(request: Request, db):
|
|
user = get_current_user(request, db=db)
|
|
if not user:
|
|
return None, RedirectResponse(url="/login", status_code=303)
|
|
if not _is_domain_admin(db, user):
|
|
return user, RedirectResponse(url="/system-settings", status_code=303)
|
|
return user, None
|
|
|
|
|
|
def _to_int(value: str | int | None) -> int | None:
|
|
try:
|
|
n = int(value or 0)
|
|
return n if n > 0 else None
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def _payload_from_form(
|
|
*,
|
|
domain_name: str,
|
|
domain_type: str,
|
|
tenant_id: str | int | None,
|
|
branch_id: str | int | None,
|
|
consultant_id: str | int | None,
|
|
parent_tenant_id: str | int | None,
|
|
is_primary: str | None,
|
|
is_verified: str | None,
|
|
status: str,
|
|
ssl_mode: str,
|
|
notes: str,
|
|
) -> DomainMappingPayload:
|
|
return DomainMappingPayload(
|
|
domain_name=domain_name,
|
|
domain_type=domain_type,
|
|
tenant_id=_to_int(tenant_id),
|
|
branch_id=_to_int(branch_id),
|
|
consultant_id=_to_int(consultant_id),
|
|
parent_tenant_id=_to_int(parent_tenant_id),
|
|
is_primary=bool(is_primary),
|
|
is_verified=bool(is_verified),
|
|
status=status or "draft",
|
|
ssl_mode=ssl_mode or "manual",
|
|
notes=notes,
|
|
)
|
|
|
|
|
|
@router.get("")
|
|
def domain_list(request: Request, q: str = "", status: str = "", domain_type: str = ""):
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user, response = _require_user(request, db)
|
|
if response:
|
|
return response
|
|
mappings = list_domain_mappings(db, q=q, status=status, domain_type=domain_type)
|
|
refs = reference_data(db)
|
|
return templates.TemplateResponse(
|
|
"modules/domain_management/templates/domain_management/list.html",
|
|
_base_ctx(
|
|
request,
|
|
db,
|
|
user,
|
|
title="Domain Mapping",
|
|
mappings=mappings,
|
|
filters={"q": q, "status": status, "domain_type": domain_type},
|
|
**refs,
|
|
),
|
|
)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.get("/new")
|
|
def domain_create_page(request: Request):
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user, response = _require_user(request, db)
|
|
if response:
|
|
return response
|
|
return templates.TemplateResponse(
|
|
"modules/domain_management/templates/domain_management/form.html",
|
|
_base_ctx(
|
|
request,
|
|
db,
|
|
user,
|
|
title="Add Domain Mapping",
|
|
mapping=None,
|
|
errors=[],
|
|
form={},
|
|
**reference_data(db),
|
|
),
|
|
)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.post("/new")
|
|
def domain_create_submit(
|
|
request: Request,
|
|
csrf_token: str = Form(...),
|
|
domain_name: str = Form(...),
|
|
domain_type: str = Form(...),
|
|
tenant_id: str = Form(""),
|
|
branch_id: str = Form(""),
|
|
consultant_id: str = Form(""),
|
|
parent_tenant_id: str = Form(""),
|
|
is_primary: str | None = Form(None),
|
|
is_verified: str | None = Form(None),
|
|
status: str = Form("draft"),
|
|
ssl_mode: str = Form("manual"),
|
|
notes: str = Form(""),
|
|
):
|
|
validate_csrf(request, csrf_token)
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user, response = _require_user(request, db)
|
|
if response:
|
|
return response
|
|
payload = _payload_from_form(
|
|
domain_name=domain_name,
|
|
domain_type=domain_type,
|
|
tenant_id=tenant_id,
|
|
branch_id=branch_id,
|
|
consultant_id=consultant_id,
|
|
parent_tenant_id=parent_tenant_id,
|
|
is_primary=is_primary,
|
|
is_verified=is_verified,
|
|
status=status,
|
|
ssl_mode=ssl_mode,
|
|
notes=notes,
|
|
)
|
|
errors = validate_domain_payload(db, payload)
|
|
if errors:
|
|
return templates.TemplateResponse(
|
|
"modules/domain_management/templates/domain_management/form.html",
|
|
_base_ctx(request, db, user, title="Add Domain Mapping", mapping=None, errors=errors, form=payload.__dict__, **reference_data(db)),
|
|
status_code=400,
|
|
)
|
|
mapping = create_domain_mapping(db, payload, user_id=user.id)
|
|
return RedirectResponse(url=f"/domains/{mapping.id}", status_code=303)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
|
|
@router.get("/tenant-subdomains")
|
|
def tenant_subdomain_page(request: Request, base_domain: str = "filingabc.com"):
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user, response = _require_user(request, db)
|
|
if response:
|
|
return response
|
|
refs = reference_data(db)
|
|
candidates = list_tenant_subdomain_candidates(db, base_domain=base_domain)
|
|
return templates.TemplateResponse(
|
|
"modules/domain_management/templates/domain_management/tenant_subdomains.html",
|
|
_base_ctx(
|
|
request,
|
|
db,
|
|
user,
|
|
title="Tenant Subdomains",
|
|
base_domain=base_domain,
|
|
candidates=candidates,
|
|
message="",
|
|
errors=[],
|
|
**refs,
|
|
),
|
|
)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.post("/tenant-subdomains/create")
|
|
def tenant_subdomain_create(
|
|
request: Request,
|
|
csrf_token: str = Form(...),
|
|
tenant_id: str = Form(...),
|
|
base_domain: str = Form("filingabc.com"),
|
|
branch_id: str = Form(""),
|
|
mark_verified_active: str | None = Form(None),
|
|
):
|
|
validate_csrf(request, csrf_token)
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user, response = _require_user(request, db)
|
|
if response:
|
|
return response
|
|
errors: list[str] = []
|
|
message = ""
|
|
try:
|
|
mapping, created = create_or_get_tenant_subdomain_mapping(
|
|
db,
|
|
tenant_id=int(tenant_id),
|
|
base_domain=base_domain,
|
|
branch_id=_to_int(branch_id),
|
|
mark_verified_active=bool(mark_verified_active),
|
|
user_id=user.id,
|
|
)
|
|
if created:
|
|
return RedirectResponse(url=f"/domains/{mapping.id}", status_code=303)
|
|
message = f"Domain mapping already exists: {mapping.domain_name}"
|
|
except Exception as exc:
|
|
errors.append(str(exc))
|
|
refs = reference_data(db)
|
|
return templates.TemplateResponse(
|
|
"modules/domain_management/templates/domain_management/tenant_subdomains.html",
|
|
_base_ctx(
|
|
request,
|
|
db,
|
|
user,
|
|
title="Tenant Subdomains",
|
|
base_domain=base_domain,
|
|
candidates=list_tenant_subdomain_candidates(db, base_domain=base_domain),
|
|
message=message,
|
|
errors=errors,
|
|
**refs,
|
|
),
|
|
status_code=400 if errors else 200,
|
|
)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.get("/firm-domain")
|
|
def firm_custom_domain_page(request: Request, domain_name: str = "arrr.associates"):
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user, response = _require_user(request, db)
|
|
if response:
|
|
return response
|
|
refs = reference_data(db)
|
|
existing = None
|
|
normalized = (domain_name or "arrr.associates").strip().lower().split("/", 1)[0].split(":", 1)[0].rstrip(".")
|
|
if normalized:
|
|
existing = db.execute(select(DomainMapping).where(DomainMapping.domain_name == normalized)).scalar_one_or_none()
|
|
return templates.TemplateResponse(
|
|
"modules/domain_management/templates/domain_management/firm_domain.html",
|
|
_base_ctx(
|
|
request,
|
|
db,
|
|
user,
|
|
title="Audit Firm Custom Domain",
|
|
domain_name=normalized or "arrr.associates",
|
|
existing=existing,
|
|
message="",
|
|
errors=[],
|
|
**refs,
|
|
),
|
|
)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.post("/firm-domain/create")
|
|
def firm_custom_domain_create(
|
|
request: Request,
|
|
csrf_token: str = Form(...),
|
|
tenant_id: str = Form(...),
|
|
domain_name: str = Form("arrr.associates"),
|
|
branch_id: str = Form(""),
|
|
mark_verified_active: str | None = Form(None),
|
|
):
|
|
validate_csrf(request, csrf_token)
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user, response = _require_user(request, db)
|
|
if response:
|
|
return response
|
|
errors: list[str] = []
|
|
message = ""
|
|
try:
|
|
mapping, created = create_or_get_firm_custom_domain_mapping(
|
|
db,
|
|
tenant_id=int(tenant_id),
|
|
domain_name=domain_name,
|
|
branch_id=_to_int(branch_id),
|
|
mark_verified_active=bool(mark_verified_active),
|
|
user_id=user.id,
|
|
)
|
|
if created:
|
|
return RedirectResponse(url=f"/domains/{mapping.id}", status_code=303)
|
|
message = f"Domain mapping already exists: {mapping.domain_name}"
|
|
except Exception as exc:
|
|
errors.append(str(exc))
|
|
refs = reference_data(db)
|
|
normalized = (domain_name or "arrr.associates").strip().lower().split("/", 1)[0].split(":", 1)[0].rstrip(".")
|
|
existing = None
|
|
if normalized:
|
|
existing = db.execute(select(DomainMapping).where(DomainMapping.domain_name == normalized)).scalar_one_or_none()
|
|
return templates.TemplateResponse(
|
|
"modules/domain_management/templates/domain_management/firm_domain.html",
|
|
_base_ctx(
|
|
request,
|
|
db,
|
|
user,
|
|
title="Audit Firm Custom Domain",
|
|
domain_name=normalized or "arrr.associates",
|
|
existing=existing,
|
|
message=message,
|
|
errors=errors,
|
|
**refs,
|
|
),
|
|
status_code=400 if errors else 200,
|
|
)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.get("/consultant-domains")
|
|
def consultant_domains_page(request: Request, base_domain: str = "filingabc.com", firm_base_domain: str = "arrr.accountant"):
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user, response = _require_user(request, db)
|
|
if response:
|
|
return response
|
|
refs = reference_data(db)
|
|
candidates = list_consultant_domain_candidates(db, base_domain=base_domain, firm_base_domain=firm_base_domain)
|
|
return templates.TemplateResponse(
|
|
"modules/domain_management/templates/domain_management/consultant_domains.html",
|
|
_base_ctx(
|
|
request,
|
|
db,
|
|
user,
|
|
title="Consultant Profile Domains",
|
|
base_domain=base_domain,
|
|
firm_base_domain=firm_base_domain,
|
|
candidates=candidates,
|
|
message="",
|
|
errors=[],
|
|
**refs,
|
|
),
|
|
)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.post("/consultant-domains/create")
|
|
def consultant_domain_create(
|
|
request: Request,
|
|
csrf_token: str = Form(...),
|
|
consultant_id: str = Form(...),
|
|
domain_type: str = Form("consultant_marketplace_subdomain"),
|
|
domain_name: str = Form(""),
|
|
base_domain: str = Form("filingabc.com"),
|
|
firm_base_domain: str = Form("arrr.accountant"),
|
|
parent_tenant_id: str = Form(""),
|
|
mark_verified_active: str | None = Form(None),
|
|
):
|
|
validate_csrf(request, csrf_token)
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user, response = _require_user(request, db)
|
|
if response:
|
|
return response
|
|
errors: list[str] = []
|
|
message = ""
|
|
try:
|
|
mapping, created = create_or_get_consultant_domain_mapping(
|
|
db,
|
|
consultant_id=int(consultant_id),
|
|
domain_type=domain_type,
|
|
domain_name=domain_name or None,
|
|
base_domain=base_domain,
|
|
firm_base_domain=firm_base_domain,
|
|
parent_tenant_id=_to_int(parent_tenant_id),
|
|
mark_verified_active=bool(mark_verified_active),
|
|
user_id=user.id,
|
|
)
|
|
if created:
|
|
return RedirectResponse(url=f"/domains/{mapping.id}", status_code=303)
|
|
message = f"Domain mapping already exists: {mapping.domain_name}"
|
|
except Exception as exc:
|
|
errors.append(str(exc))
|
|
refs = reference_data(db)
|
|
return templates.TemplateResponse(
|
|
"modules/domain_management/templates/domain_management/consultant_domains.html",
|
|
_base_ctx(
|
|
request,
|
|
db,
|
|
user,
|
|
title="Consultant Profile Domains",
|
|
base_domain=base_domain,
|
|
firm_base_domain=firm_base_domain,
|
|
candidates=list_consultant_domain_candidates(db, base_domain=base_domain, firm_base_domain=firm_base_domain),
|
|
message=message,
|
|
errors=errors,
|
|
**refs,
|
|
),
|
|
status_code=400 if errors else 200,
|
|
)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.get("/verification")
|
|
def domain_verification_page(request: Request):
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user, response = _require_user(request, db)
|
|
if response:
|
|
return response
|
|
pending = list_domains_requiring_verification(db)
|
|
return templates.TemplateResponse(
|
|
"modules/domain_management/templates/domain_management/verification.html",
|
|
_base_ctx(
|
|
request,
|
|
db,
|
|
user,
|
|
title="Custom Domain Verification",
|
|
pending=pending,
|
|
results=[],
|
|
),
|
|
)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.post("/verification/run")
|
|
def domain_verification_run(request: Request, csrf_token: str = Form(...), limit: str = Form("25")):
|
|
validate_csrf(request, csrf_token)
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user, response = _require_user(request, db)
|
|
if response:
|
|
return response
|
|
try:
|
|
n = int(limit or 25)
|
|
except Exception:
|
|
n = 25
|
|
results = verify_all_pending_domains(db, user_id=user.id, limit=n)
|
|
return templates.TemplateResponse(
|
|
"modules/domain_management/templates/domain_management/verification.html",
|
|
_base_ctx(
|
|
request,
|
|
db,
|
|
user,
|
|
title="Custom Domain Verification",
|
|
pending=list_domains_requiring_verification(db),
|
|
results=results,
|
|
),
|
|
)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.get("/ssl")
|
|
def ssl_dashboard(request: Request):
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user, response = _require_user(request, db)
|
|
if response:
|
|
return response
|
|
domains = list_ssl_domains(db)
|
|
return templates.TemplateResponse(
|
|
"modules/domain_management/templates/domain_management/ssl.html",
|
|
_base_ctx(request, db, user, title="SSL Automation", domains=domains),
|
|
)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.post("/ssl/check-all")
|
|
def ssl_check_all(request: Request, csrf_token: str = Form(...)):
|
|
validate_csrf(request, csrf_token)
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user, response = _require_user(request, db)
|
|
if response:
|
|
return response
|
|
results = []
|
|
for mapping in list_ssl_domains(db)[:25]:
|
|
if mapping.status == "active" and mapping.is_verified:
|
|
result = check_domain_ssl_certificate(db, mapping, user_id=user.id)
|
|
results.append((mapping, result))
|
|
domains = list_ssl_domains(db)
|
|
return templates.TemplateResponse(
|
|
"modules/domain_management/templates/domain_management/ssl.html",
|
|
_base_ctx(request, db, user, title="SSL Automation", domains=domains, ssl_results=results),
|
|
)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.get("/coolify/status")
|
|
def coolify_status_page(request: Request):
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user, response = _require_user(request, db)
|
|
if response:
|
|
return response
|
|
result = get_coolify_application_domains()
|
|
domains = list_ssl_domains(db)
|
|
return templates.TemplateResponse(
|
|
"modules/domain_management/templates/domain_management/ssl.html",
|
|
_base_ctx(request, db, user, title="Coolify Domain Sync", domains=domains, coolify_result=result),
|
|
status_code=200 if result.ok else 400,
|
|
)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.get("/{mapping_id}/ssl")
|
|
def ssl_detail(request: Request, mapping_id: int):
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user, response = _require_user(request, db)
|
|
if response:
|
|
return response
|
|
mapping = db.execute(select(DomainMapping).where(DomainMapping.id == mapping_id)).scalar_one_or_none()
|
|
if not mapping:
|
|
return RedirectResponse(url="/domains/ssl", status_code=303)
|
|
snippets = build_ssl_proxy_snippets(mapping)
|
|
return templates.TemplateResponse(
|
|
"modules/domain_management/templates/domain_management/ssl_detail.html",
|
|
_base_ctx(request, db, user, title=f"SSL: {mapping.domain_name}", mapping=mapping, snippets=snippets),
|
|
)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.post("/{mapping_id}/coolify/sync")
|
|
def coolify_sync_one(request: Request, mapping_id: int, csrf_token: str = Form(...)):
|
|
validate_csrf(request, csrf_token)
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user, response = _require_user(request, db)
|
|
if response:
|
|
return response
|
|
mapping = db.execute(select(DomainMapping).where(DomainMapping.id == mapping_id)).scalar_one_or_none()
|
|
if not mapping:
|
|
return RedirectResponse(url="/domains", status_code=303)
|
|
result = sync_domain_to_coolify(db, mapping, user_id=user.id)
|
|
db.refresh(mapping)
|
|
return templates.TemplateResponse(
|
|
"modules/domain_management/templates/domain_management/detail.html",
|
|
_base_ctx(request, db, user, title=mapping.domain_name, mapping=mapping, coolify_result=result),
|
|
status_code=200 if result.ok else 400,
|
|
)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.post("/{mapping_id}/coolify/check")
|
|
def coolify_check_one(request: Request, mapping_id: int, csrf_token: str = Form(...)):
|
|
validate_csrf(request, csrf_token)
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user, response = _require_user(request, db)
|
|
if response:
|
|
return response
|
|
mapping = db.execute(select(DomainMapping).where(DomainMapping.id == mapping_id)).scalar_one_or_none()
|
|
if not mapping:
|
|
return RedirectResponse(url="/domains", status_code=303)
|
|
result = get_coolify_application_domains()
|
|
return templates.TemplateResponse(
|
|
"modules/domain_management/templates/domain_management/detail.html",
|
|
_base_ctx(request, db, user, title=mapping.domain_name, mapping=mapping, coolify_result=result),
|
|
status_code=200 if result.ok else 400,
|
|
)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.post("/{mapping_id}/ssl/check")
|
|
def ssl_check_one(request: Request, mapping_id: int, csrf_token: str = Form(...)):
|
|
validate_csrf(request, csrf_token)
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user, response = _require_user(request, db)
|
|
if response:
|
|
return response
|
|
mapping = db.execute(select(DomainMapping).where(DomainMapping.id == mapping_id)).scalar_one_or_none()
|
|
if not mapping:
|
|
return RedirectResponse(url="/domains/ssl", status_code=303)
|
|
result = check_domain_ssl_certificate(db, mapping, user_id=user.id)
|
|
snippets = build_ssl_proxy_snippets(mapping)
|
|
return templates.TemplateResponse(
|
|
"modules/domain_management/templates/domain_management/ssl_detail.html",
|
|
_base_ctx(request, db, user, title=f"SSL: {mapping.domain_name}", mapping=mapping, snippets=snippets, ssl_result=result),
|
|
status_code=200 if result.ok else 400,
|
|
)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.post("/{mapping_id}/ssl/mark-managed")
|
|
def ssl_mark_managed(request: Request, mapping_id: int, csrf_token: str = Form(...), provider: str = Form("manual")):
|
|
validate_csrf(request, csrf_token)
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user, response = _require_user(request, db)
|
|
if response:
|
|
return response
|
|
mapping = db.execute(select(DomainMapping).where(DomainMapping.id == mapping_id)).scalar_one_or_none()
|
|
if mapping:
|
|
mark_ssl_managed(db, mapping, provider=provider, user_id=user.id)
|
|
return RedirectResponse(url=f"/domains/{mapping_id}/ssl", status_code=303)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.get("/{mapping_id}")
|
|
def domain_detail(request: Request, mapping_id: int):
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user, response = _require_user(request, db)
|
|
if response:
|
|
return response
|
|
mapping = db.execute(select(DomainMapping).where(DomainMapping.id == mapping_id)).scalar_one_or_none()
|
|
if not mapping:
|
|
return RedirectResponse(url="/domains", status_code=303)
|
|
return templates.TemplateResponse(
|
|
"modules/domain_management/templates/domain_management/detail.html",
|
|
_base_ctx(request, db, user, title=mapping.domain_name, mapping=mapping),
|
|
)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.get("/{mapping_id}/edit")
|
|
def domain_edit_page(request: Request, mapping_id: int):
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user, response = _require_user(request, db)
|
|
if response:
|
|
return response
|
|
mapping = db.execute(select(DomainMapping).where(DomainMapping.id == mapping_id)).scalar_one_or_none()
|
|
if not mapping:
|
|
return RedirectResponse(url="/domains", status_code=303)
|
|
return templates.TemplateResponse(
|
|
"modules/domain_management/templates/domain_management/form.html",
|
|
_base_ctx(request, db, user, title="Edit Domain Mapping", mapping=mapping, errors=[], form={}, **reference_data(db)),
|
|
)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.post("/{mapping_id}/edit")
|
|
def domain_edit_submit(
|
|
request: Request,
|
|
mapping_id: int,
|
|
csrf_token: str = Form(...),
|
|
domain_name: str = Form(...),
|
|
domain_type: str = Form(...),
|
|
tenant_id: str = Form(""),
|
|
branch_id: str = Form(""),
|
|
consultant_id: str = Form(""),
|
|
parent_tenant_id: str = Form(""),
|
|
is_primary: str | None = Form(None),
|
|
is_verified: str | None = Form(None),
|
|
status: str = Form("draft"),
|
|
ssl_mode: str = Form("manual"),
|
|
notes: str = Form(""),
|
|
):
|
|
validate_csrf(request, csrf_token)
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user, response = _require_user(request, db)
|
|
if response:
|
|
return response
|
|
mapping = db.execute(select(DomainMapping).where(DomainMapping.id == mapping_id)).scalar_one_or_none()
|
|
if not mapping:
|
|
return RedirectResponse(url="/domains", status_code=303)
|
|
payload = _payload_from_form(
|
|
domain_name=domain_name,
|
|
domain_type=domain_type,
|
|
tenant_id=tenant_id,
|
|
branch_id=branch_id,
|
|
consultant_id=consultant_id,
|
|
parent_tenant_id=parent_tenant_id,
|
|
is_primary=is_primary,
|
|
is_verified=is_verified,
|
|
status=status,
|
|
ssl_mode=ssl_mode,
|
|
notes=notes,
|
|
)
|
|
errors = validate_domain_payload(db, payload, mapping_id=mapping.id)
|
|
if errors:
|
|
return templates.TemplateResponse(
|
|
"modules/domain_management/templates/domain_management/form.html",
|
|
_base_ctx(request, db, user, title="Edit Domain Mapping", mapping=mapping, errors=errors, form=payload.__dict__, **reference_data(db)),
|
|
status_code=400,
|
|
)
|
|
update_domain_mapping(db, mapping, payload, user_id=user.id)
|
|
return RedirectResponse(url=f"/domains/{mapping.id}", status_code=303)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.post("/{mapping_id}/verify-dns")
|
|
def domain_verify_dns(request: Request, mapping_id: int, csrf_token: str = Form(...)):
|
|
validate_csrf(request, csrf_token)
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user, response = _require_user(request, db)
|
|
if response:
|
|
return response
|
|
mapping = db.execute(select(DomainMapping).where(DomainMapping.id == mapping_id)).scalar_one_or_none()
|
|
if not mapping:
|
|
return RedirectResponse(url="/domains", status_code=303)
|
|
result = verify_domain_dns_txt(db, mapping, user_id=user.id)
|
|
return templates.TemplateResponse(
|
|
"modules/domain_management/templates/domain_management/detail.html",
|
|
_base_ctx(request, db, user, title=mapping.domain_name, mapping=mapping, dns_result=result),
|
|
status_code=200 if result.ok else 400,
|
|
)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.post("/{mapping_id}/regenerate-token")
|
|
def domain_regenerate_token(request: Request, mapping_id: int, csrf_token: str = Form(...)):
|
|
validate_csrf(request, csrf_token)
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user, response = _require_user(request, db)
|
|
if response:
|
|
return response
|
|
mapping = db.execute(select(DomainMapping).where(DomainMapping.id == mapping_id)).scalar_one_or_none()
|
|
if mapping:
|
|
regenerate_verification_token(db, mapping, user_id=user.id)
|
|
return RedirectResponse(url=f"/domains/{mapping_id}", status_code=303)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.post("/{mapping_id}/mark-verified")
|
|
def domain_mark_verified(request: Request, mapping_id: int, csrf_token: str = Form(...)):
|
|
validate_csrf(request, csrf_token)
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user, response = _require_user(request, db)
|
|
if response:
|
|
return response
|
|
mapping = db.execute(select(DomainMapping).where(DomainMapping.id == mapping_id)).scalar_one_or_none()
|
|
if mapping:
|
|
mark_verified(db, mapping, user_id=user.id)
|
|
return RedirectResponse(url=f"/domains/{mapping_id}", status_code=303)
|
|
finally:
|
|
db.close()
|