Add Obin AD user creator scaffold

obin-ad-user-creator-compose.yml

@@ -0,0 +1,31 @@
+services:
+  obin-ad-user-creator:
+    build:
+      context: ./obin-ad-user-creator
+      dockerfile: Dockerfile
+    container_name: obin-ad-user-creator
+    restart: unless-stopped
+    environment:
+      APP_TITLE: ${APP_TITLE:-Wheelytho Account Creation}
+      MINIMUM_PASSWORD_LENGTH: ${MINIMUM_PASSWORD_LENGTH:-12}
+      APP_REQUIRE_INVITE_CODE: ${APP_REQUIRE_INVITE_CODE:-false}
+      APP_INVITE_CODE: ${APP_INVITE_CODE-}
+      WELCOME_URL: ${WELCOME_URL:-https://welcome.wheelytho.com/}
+      AUTHELIA_SETUP_URL: ${AUTHELIA_SETUP_URL:-https://auth.wheelytho.com/}
+      AD_LDAPS_URL: ${AD_LDAPS_URL:-ldaps://192.168.30.15:636}
+      AD_BIND_USERNAME: ${AD_BIND_USERNAME:?set AD_BIND_USERNAME in Portainer runtime env}
+      AD_BIND_PASSWORD: ${AD_BIND_PASSWORD:?set AD_BIND_PASSWORD in Portainer runtime env}
+      AD_BASE_DN: ${AD_BASE_DN:-DC=local,DC=wheelz,DC=com}
+      AD_CREATE_OU: ${AD_CREATE_OU:-OU=WheelzUsers,DC=local,DC=wheelz,DC=com}
+      AD_DEFAULT_GROUPS: ${AD_DEFAULT_GROUPS:-Authelia-Friends}
+      AD_DOMAIN_UPN_SUFFIX: ${AD_DOMAIN_UPN_SUFFIX:-local.wheelz.com}
+      # Current lab AD cert is not trusted by the container yet. Move this to true after CA trust is added.
+      AD_TLS_VALIDATE: ${AD_TLS_VALIDATE:-false}
+    ports:
+      - "${OBIN_AD_USER_CREATOR_PORT:-8098}:8080"
+    healthcheck:
+      test: ["CMD", "python", "-c", "import urllib.request; urllib.request.urlopen('http://127.0.0.1:8080/health', timeout=3)"]
+      interval: 30s
+      timeout: 5s
+      retries: 3
+      start_period: 10s

obin-ad-user-creator.env.example

@@ -0,0 +1,27 @@
+# Obin AD User Creator runtime settings
+# Copy real values into Portainer or /home/wheelz/.hermes/runtime-secrets/obin-ad-user-creator.env.
+# Do not commit real passwords or invite codes.
+
+APP_TITLE=Wheelytho Account Creation
+MINIMUM_PASSWORD_LENGTH=12
+# Keep this enabled for any public account-creation page.
+APP_REQUIRE_INVITE_CODE=true
+APP_INVITE_CODE=
+
+WELCOME_URL=https://welcome.wheelytho.com/
+AUTHELIA_SETUP_URL=https://auth.wheelytho.com/
+
+# AD connection. Password set/reset requires LDAPS or StartTLS.
+AD_LDAPS_URL=ldaps://192.168.30.15:636
+AD_BIND_USERNAME=
+AD_BIND_PASSWORD=
+AD_BASE_DN=DC=local,DC=wheelz,DC=com
+AD_CREATE_OU=OU=WheelzUsers,DC=local,DC=wheelz,DC=com
+AD_DEFAULT_GROUPS=Authelia-Friends
+AD_DOMAIN_UPN_SUFFIX=local.wheelz.com
+
+# Temporarily false because the container does not yet trust the AD CA/root cert.
+# Later hardening: add the AD CA cert and set true.
+AD_TLS_VALIDATE=false
+
+OBIN_AD_USER_CREATOR_PORT=8098

obin-ad-user-creator/.gitignore

@@ -0,0 +1,6 @@
+__pycache__/
+.pytest_cache/
+.venv/
+*.py[cod]
+.env
+*.env

obin-ad-user-creator/Dockerfile

@@ -0,0 +1,12 @@
+FROM python:3.12-slim
+
+ENV PYTHONDONTWRITEBYTECODE=1 \
+    PYTHONUNBUFFERED=1
+
+WORKDIR /app
+COPY requirements.txt /app/requirements.txt
+RUN pip install --no-cache-dir -r /app/requirements.txt
+COPY app /app/app
+
+EXPOSE 8080
+CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8080"]

obin-ad-user-creator/README.md

@@ -0,0 +1,24 @@
+# Obin AD User Creator
+
+Small FastAPI app for creating Wheelytho Active Directory accounts from a web form.
+
+Current scope:
+- Collect username, password, and email.
+- Create an AD user in one configured OU.
+- Set the initial password over LDAPS.
+- Enable the account.
+- Add the user to configured default Authelia groups, currently intended as `Authelia-Friends`.
+- Redirect to a created/instructions page with links to Authelia and the future welcome page.
+
+Security notes:
+- Use a dedicated AD creator service account, not the existing Authelia read/bind account.
+- Delegate the service account only to the target OU and required default group membership.
+- Password setting requires LDAPS/SSL to AD. Do not run AD password creation over plain LDAP.
+- Keep real AD creator credentials and invite codes in Portainer runtime env or `/home/wheelz/.hermes/runtime-secrets/obin-ad-user-creator.env`, not in Gitea or Obsidian.
+- The first deployment should be internal-only until the OU, group, and rollback behavior are verified.
+- If exposed externally later, keep `APP_REQUIRE_INVITE_CODE=true` unless the page is protected by a separate approval/admin workflow.
+
+Validation already available:
+- `pytest -q`
+- `docker compose --env-file <runtime.env> -f obin-ad-user-creator-compose.yml config`
+- `GET /health`

obin-ad-user-creator/app/main.py

@@ -0,0 +1,263 @@
+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>
+        <div class='notice'>Password minimum: {settings.minimum_password_length} characters. Your account will be added to the configured AD OU and default access group.</div>
+    """)
+
+
+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>
+    """)

obin-ad-user-creator/requirements.txt

@@ -0,0 +1,5 @@
+fastapi==0.115.6
+uvicorn[standard]==0.34.0
+ldap3==2.9.1
+python-multipart==0.0.20
+httpx==0.28.1

obin-ad-user-creator/tests/test_app.py

@@ -0,0 +1,110 @@
+import os
+
+import pytest
+from fastapi.testclient import TestClient
+
+os.environ.update({
+    "APP_TITLE": "Wheelytho Account Creation",
+    "AD_LDAPS_URL": "ldaps://dc.example.test:636",
+    "AD_BIND_USERNAME": "svc-create@example.test",
+    "AD_BIND_PASSWORD": "secret",
+    "AD_BASE_DN": "DC=example,DC=test",
+    "AD_CREATE_OU": "OU=WheelzUsers,DC=example,DC=test",
+    "AD_DEFAULT_GROUPS": "Authelia-Family",
+    "AD_DOMAIN_UPN_SUFFIX": "example.test",
+    "APP_REQUIRE_INVITE_CODE": "false",
+    "APP_INVITE_CODE": "",
+    "WELCOME_URL": "https://welcome.wheelytho.com/",
+    "AUTHELIA_SETUP_URL": "https://auth.wheelytho.com/",
+})
+
+from app.main import app, get_ad_client
+
+
+class FakeADClient:
+    def __init__(self):
+        self.created = []
+
+    def create_user(self, username, password, email):
+        self.created.append({"username": username, "password": password, "email": email})
+        return {"username": username, "email": email, "groups": ["Authelia-Family"]}
+
+
+@pytest.fixture()
+def fake_client(monkeypatch):
+    monkeypatch.setenv("APP_REQUIRE_INVITE_CODE", "false")
+    fake = FakeADClient()
+    app.dependency_overrides[get_ad_client] = lambda: fake
+    yield TestClient(app, follow_redirects=False), fake
+    app.dependency_overrides.clear()
+
+
+def test_form_loads():
+    client = TestClient(app)
+    response = client.get("/")
+    assert response.status_code == 200
+    assert "Wheelytho Account Creation" in response.text
+    assert "name='username'" in response.text
+    assert "name='email'" in response.text
+    assert "name='password'" in response.text
+
+
+def test_create_user_success_returns_redirect_and_never_echoes_password(fake_client):
+    client, fake = fake_client
+    response = client.post("/api/create-user", data={
+        "username": "new.family",
+        "password": "CorrectHorseBatteryStaple!42",
+        "email": "New.Family@Example.com",
+    })
+    assert response.status_code == 303
+    assert response.headers["location"].startswith("/created?username=new.family")
+    assert fake.created == [{"username": "new.family", "password": "CorrectHorseBatteryStaple!42", "email": "new.family@example.com"}]
+    assert "CorrectHorseBatteryStaple" not in response.text
+
+
+def test_create_user_rejects_unsafe_username_before_ad_call(fake_client):
+    client, fake = fake_client
+    response = client.post("/api/create-user", data={
+        "username": "../../admin",
+        "password": "CorrectHorseBatteryStaple!42",
+        "email": "bad@example.com",
+    })
+    assert response.status_code == 400
+    assert fake.created == []
+
+
+def test_create_user_rejects_weak_password_before_ad_call(fake_client):
+    client, fake = fake_client
+    response = client.post("/api/create-user", data={
+        "username": "newuser",
+        "password": "short",
+        "email": "newuser@example.com",
+    })
+    assert response.status_code == 400
+    assert fake.created == []
+
+
+def test_create_user_requires_invite_code_when_enabled(monkeypatch):
+    monkeypatch.setenv("APP_REQUIRE_INVITE_CODE", "true")
+    monkeypatch.setenv("APP_INVITE_CODE", "family-only")
+    fake = FakeADClient()
+    app.dependency_overrides[get_ad_client] = lambda: fake
+    client = TestClient(app, follow_redirects=False)
+    response = client.post("/api/create-user", data={
+        "username": "newuser",
+        "password": "CorrectHorseBatteryStaple!42",
+        "email": "newuser@example.com",
+        "invite_code": "wrong",
+    })
+    assert response.status_code == 403
+    assert fake.created == []
+    app.dependency_overrides.clear()
+
+
+def test_created_page_contains_2fa_instructions_and_welcome_link():
+    client = TestClient(app)
+    response = client.get("/created?username=newuser&email=newuser%40example.com")
+    assert response.status_code == 200
+    assert "Set up 2FA" in response.text
+    assert "https://auth.wheelytho.com/" in response.text
+    assert "https://welcome.wheelytho.com/" in response.text