263 lines
12 KiB
Python
263 lines
12 KiB
Python
import os
|
|
import re
|
|
import ssl
|
|
from dataclasses import dataclass
|
|
from html import escape
|
|
from typing import List, Optional
|
|
from urllib.parse import quote, urlparse
|
|
|
|
from fastapi import Depends, FastAPI, Form, HTTPException, Request
|
|
from fastapi.responses import HTMLResponse, RedirectResponse
|
|
|
|
USERNAME_RE = re.compile(r"^[A-Za-z][A-Za-z0-9._-]{2,31}$")
|
|
|
|
|
|
def env(name: str, default: Optional[str] = None, required: bool = False) -> str:
|
|
value = os.getenv(name, default)
|
|
if required and not value:
|
|
raise RuntimeError(f"Missing required environment variable: {name}")
|
|
return value or ""
|
|
|
|
|
|
def env_bool(name: str, default: bool = False) -> bool:
|
|
return env(name, "true" if default else "false").strip().lower() in {"1", "true", "yes", "on"}
|
|
|
|
|
|
def split_csv(value: str) -> List[str]:
|
|
return [item.strip() for item in value.split(",") if item.strip()]
|
|
|
|
|
|
@dataclass
|
|
class Settings:
|
|
app_title: str
|
|
minimum_password_length: int
|
|
welcome_url: str
|
|
authelia_setup_url: str
|
|
require_invite_code: bool
|
|
invite_code: str
|
|
ad_ldaps_url: str
|
|
ad_bind_username: str
|
|
ad_bind_password: str
|
|
ad_base_dn: str
|
|
ad_create_ou: str
|
|
ad_default_groups: List[str]
|
|
ad_upn_suffix: str
|
|
ad_tls_validate: bool
|
|
|
|
@classmethod
|
|
def from_env(cls) -> "Settings":
|
|
return cls(
|
|
app_title=env("APP_TITLE", "Obin Account Creation"),
|
|
minimum_password_length=int(env("MINIMUM_PASSWORD_LENGTH", "12")),
|
|
welcome_url=env("WELCOME_URL", "https://welcome.wheelytho.com/"),
|
|
authelia_setup_url=env("AUTHELIA_SETUP_URL", "https://auth.wheelytho.com/"),
|
|
require_invite_code=env_bool("APP_REQUIRE_INVITE_CODE", False),
|
|
invite_code=env("APP_INVITE_CODE", ""),
|
|
ad_ldaps_url=env("AD_LDAPS_URL", required=True),
|
|
ad_bind_username=env("AD_BIND_USERNAME", required=True),
|
|
ad_bind_password=env("AD_BIND_PASSWORD", required=True),
|
|
ad_base_dn=env("AD_BASE_DN", required=True),
|
|
ad_create_ou=env("AD_CREATE_OU", required=True),
|
|
ad_default_groups=split_csv(env("AD_DEFAULT_GROUPS", "Authelia-Family")),
|
|
ad_upn_suffix=env("AD_DOMAIN_UPN_SUFFIX", "local.wheelz.com"),
|
|
ad_tls_validate=env_bool("AD_TLS_VALIDATE", True),
|
|
)
|
|
|
|
|
|
class ADUserCreator:
|
|
def __init__(self, settings: Settings):
|
|
self.settings = settings
|
|
|
|
def create_user(self, username: str, password: str, email: str) -> dict:
|
|
from ldap3 import ALL, MODIFY_ADD, MODIFY_REPLACE, Connection, Server, Tls
|
|
from ldap3.core.exceptions import LDAPException
|
|
from ldap3.utils.dn import escape_rdn
|
|
|
|
target = self.settings.ad_ldaps_url
|
|
parsed = urlparse(target if "://" in target else f"ldaps://{target}")
|
|
host = parsed.hostname or target
|
|
port = parsed.port or 636
|
|
use_ssl = parsed.scheme == "ldaps"
|
|
tls = Tls(validate=ssl.CERT_REQUIRED if self.settings.ad_tls_validate else ssl.CERT_NONE)
|
|
server = Server(host, port=port, use_ssl=use_ssl, tls=tls, get_info=ALL)
|
|
conn = None
|
|
user_dn = f"CN={escape_rdn(username)},{self.settings.ad_create_ou}"
|
|
try:
|
|
conn = Connection(
|
|
server,
|
|
user=self.settings.ad_bind_username,
|
|
password=self.settings.ad_bind_password,
|
|
auto_bind=True,
|
|
)
|
|
conn.search(self.settings.ad_base_dn, f"(sAMAccountName={username})", attributes=["distinguishedName"])
|
|
if conn.entries:
|
|
raise ValueError("That username already exists.")
|
|
|
|
upn = f"{username}@{self.settings.ad_upn_suffix}"
|
|
attributes = {
|
|
"objectClass": ["top", "person", "organizationalPerson", "user"],
|
|
"cn": username,
|
|
"sAMAccountName": username,
|
|
"userPrincipalName": upn,
|
|
"displayName": username,
|
|
"mail": email,
|
|
"userAccountControl": 544,
|
|
}
|
|
if not conn.add(user_dn, attributes=attributes):
|
|
raise RuntimeError(f"AD user add failed: {conn.result.get('description')} {conn.result.get('message')}")
|
|
|
|
quoted_password = f'"{password}"'.encode("utf-16-le")
|
|
if not conn.modify(user_dn, {"unicodePwd": [(MODIFY_REPLACE, [quoted_password])]}):
|
|
conn.delete(user_dn)
|
|
raise RuntimeError(f"AD password set failed; rolled back user: {conn.result.get('description')} {conn.result.get('message')}")
|
|
|
|
if not conn.modify(user_dn, {"userAccountControl": [(MODIFY_REPLACE, [512])]}):
|
|
raise RuntimeError(f"AD user enable failed: {conn.result.get('description')} {conn.result.get('message')}")
|
|
|
|
added_groups = []
|
|
for group_cn in self.settings.ad_default_groups:
|
|
conn.search(self.settings.ad_base_dn, f"(&(objectClass=group)(cn={group_cn}))", attributes=["distinguishedName"])
|
|
if not conn.entries:
|
|
raise RuntimeError(f"Default AD group not found: {group_cn}")
|
|
group_dn = str(conn.entries[0].distinguishedName)
|
|
if not conn.modify(group_dn, {"member": [(MODIFY_ADD, [user_dn])]}):
|
|
raise RuntimeError(f"AD group add failed for {group_cn}: {conn.result.get('description')} {conn.result.get('message')}")
|
|
added_groups.append(group_cn)
|
|
|
|
return {"username": username, "email": email, "dn": user_dn, "groups": added_groups}
|
|
except LDAPException as exc:
|
|
raise RuntimeError(f"LDAP operation failed: {exc}") from exc
|
|
finally:
|
|
if conn is not None:
|
|
try:
|
|
conn.unbind()
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
def get_settings() -> Settings:
|
|
return Settings.from_env()
|
|
|
|
|
|
def get_ad_client() -> ADUserCreator:
|
|
return ADUserCreator(get_settings())
|
|
|
|
|
|
app = FastAPI(title="Obin Account Creation")
|
|
|
|
BASE_CSS = """
|
|
:root { color-scheme: dark; font-family: Inter, system-ui, -apple-system, Segoe UI, sans-serif; background: #0c0f14; color: #f4efe5; }
|
|
body { margin: 0; min-height: 100vh; display: grid; place-items: center; background: radial-gradient(circle at top, #1f2937 0, #0c0f14 45%, #06070a 100%); }
|
|
main { width: min(92vw, 560px); background: rgba(19, 24, 33, .92); border: 1px solid rgba(245, 178, 81, .25); border-radius: 24px; padding: 28px; box-shadow: 0 24px 80px rgba(0,0,0,.45); }
|
|
h1 { margin: 0 0 8px; font-size: 1.8rem; }
|
|
p { color: #b9c0cc; line-height: 1.5; }
|
|
label { display: block; margin: 18px 0 7px; color: #d9d2c5; font-weight: 700; }
|
|
input { width: 100%; box-sizing: border-box; border: 1px solid #394252; background: #080b10; color: #fff7eb; border-radius: 12px; padding: 13px 14px; font-size: 1rem; }
|
|
button, .button { margin-top: 22px; display: inline-block; border: 0; background: linear-gradient(135deg, #f59e0b, #fb6b28); color: #111; font-weight: 800; border-radius: 14px; padding: 13px 18px; cursor: pointer; text-decoration: none; }
|
|
.notice { margin-top: 16px; padding: 12px 14px; background: rgba(245, 158, 11, .1); border: 1px solid rgba(245, 158, 11, .25); border-radius: 14px; color: #f8d99f; }
|
|
.error { background: rgba(239, 68, 68, .12); border-color: rgba(239, 68, 68, .35); color: #fecaca; }
|
|
ul { color: #c9d1dc; line-height: 1.7; }
|
|
"""
|
|
|
|
|
|
def page(title: str, body: str) -> HTMLResponse:
|
|
return HTMLResponse(f"""<!doctype html><html><head><meta charset='utf-8'><meta name='viewport' content='width=device-width,initial-scale=1'><title>{escape(title)}</title><style>{BASE_CSS}</style></head><body><main>{body}</main></body></html>""")
|
|
|
|
|
|
@app.get("/health")
|
|
def health():
|
|
return {"status": "ok"}
|
|
|
|
|
|
@app.get("/", response_class=HTMLResponse)
|
|
def form(request: Request):
|
|
settings = get_settings()
|
|
error = request.query_params.get("error", "")
|
|
error_html = f"<div class='notice error'>{escape(error)}</div>" if error else ""
|
|
invite_html = ""
|
|
if settings.require_invite_code:
|
|
invite_html = """
|
|
<label for='invite_code'>Invite code</label>
|
|
<input id='invite_code' name='invite_code' type='password' required autocomplete='one-time-code'>
|
|
"""
|
|
return page(settings.app_title, f"""
|
|
<h1>{escape(settings.app_title)}</h1>
|
|
<p>Create a Wheelytho account. After the account is created, you will be sent to setup instructions.</p>
|
|
{error_html}
|
|
<form method='post' action='/create' autocomplete='off'>
|
|
<label for='username'>Username</label>
|
|
<input id='username' name='username' required pattern='[A-Za-z][A-Za-z0-9._-]{{2,31}}' placeholder='firstlast'>
|
|
<label for='email'>Email</label>
|
|
<input id='email' name='email' type='email' required placeholder='you@example.com'>
|
|
<label for='password'>Password</label>
|
|
<input id='password' name='password' type='password' required minlength='{settings.minimum_password_length}' autocomplete='new-password'>
|
|
{invite_html}
|
|
<button type='submit'>Create account</button>
|
|
</form>
|
|
""")
|
|
|
|
|
|
def validate_submission(username: str, password: str, email: str, invite_code: str, settings: Settings) -> None:
|
|
if not USERNAME_RE.match(username):
|
|
raise HTTPException(status_code=400, detail="Username must start with a letter and use 3-32 letters, numbers, dots, underscores, or dashes.")
|
|
if len(password) < settings.minimum_password_length:
|
|
raise HTTPException(status_code=400, detail=f"Password must be at least {settings.minimum_password_length} characters.")
|
|
if "@" not in email or "." not in email.split("@")[-1]:
|
|
raise HTTPException(status_code=400, detail="Enter a valid email address.")
|
|
if settings.require_invite_code and invite_code != settings.invite_code:
|
|
raise HTTPException(status_code=403, detail="Invalid invite code.")
|
|
|
|
|
|
def create_account(username: str, password: str, email: str, invite_code: str, ad_client: ADUserCreator):
|
|
settings = get_settings()
|
|
username = username.strip()
|
|
email = email.strip().lower()
|
|
validate_submission(username, password, email, invite_code, settings)
|
|
try:
|
|
ad_client.create_user(username, password, email)
|
|
except ValueError as exc:
|
|
raise HTTPException(status_code=409, detail=str(exc))
|
|
except Exception as exc:
|
|
raise HTTPException(status_code=500, detail=str(exc))
|
|
return RedirectResponse(f"/created?username={quote(username)}&email={quote(email)}", status_code=303)
|
|
|
|
|
|
@app.post("/api/create-user")
|
|
def create_user_api(
|
|
username: str = Form(...),
|
|
password: str = Form(...),
|
|
email: str = Form(...),
|
|
invite_code: str = Form(""),
|
|
ad_client: ADUserCreator = Depends(get_ad_client),
|
|
):
|
|
return create_account(username, password, email, invite_code, ad_client)
|
|
|
|
|
|
@app.post("/create")
|
|
def create_user_form(username: str = Form(...), password: str = Form(...), email: str = Form(...), invite_code: str = Form("")):
|
|
try:
|
|
return create_account(username, password, email, invite_code, get_ad_client())
|
|
except HTTPException as exc:
|
|
return RedirectResponse(f"/?error={quote(str(exc.detail))}", status_code=303)
|
|
|
|
|
|
@app.get("/created", response_class=HTMLResponse)
|
|
def created(username: str = "", email: str = ""):
|
|
settings = get_settings()
|
|
safe_user = escape(username)
|
|
safe_email = escape(email)
|
|
return page("Account created", f"""
|
|
<h1>Account created</h1>
|
|
<p>Your account <strong>{safe_user}</strong> was created{f' for <strong>{safe_email}</strong>' if safe_email else ''}.</p>
|
|
<h2>Set up 2FA</h2>
|
|
<ul>
|
|
<li>Open the welcome page and sign in with the username and password you just created.</li>
|
|
<li>Because the welcome page is protected by Authelia, first login will guide you through 2FA setup.</li>
|
|
<li>Check your email for the Authelia identity verification code.</li>
|
|
<li>Register your authenticator app when prompted.</li>
|
|
<li>Save your recovery codes somewhere safe.</li>
|
|
</ul>
|
|
<p><a class='button' href='{escape(settings.welcome_url)}'>Open Wheelytho welcome page</a></p>
|
|
<p><a href='{escape(settings.authelia_setup_url)}'>Open Authelia directly</a></p>
|
|
""")
|