diff --git a/obin-ad-user-creator-compose.yml b/obin-ad-user-creator-compose.yml new file mode 100644 index 0000000..893c853 --- /dev/null +++ b/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 diff --git a/obin-ad-user-creator.env.example b/obin-ad-user-creator.env.example new file mode 100644 index 0000000..18d9332 --- /dev/null +++ b/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 diff --git a/obin-ad-user-creator/.gitignore b/obin-ad-user-creator/.gitignore new file mode 100644 index 0000000..1dc74c2 --- /dev/null +++ b/obin-ad-user-creator/.gitignore @@ -0,0 +1,6 @@ +__pycache__/ +.pytest_cache/ +.venv/ +*.py[cod] +.env +*.env diff --git a/obin-ad-user-creator/Dockerfile b/obin-ad-user-creator/Dockerfile new file mode 100644 index 0000000..bb072db --- /dev/null +++ b/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"] diff --git a/obin-ad-user-creator/README.md b/obin-ad-user-creator/README.md new file mode 100644 index 0000000..e7be26c --- /dev/null +++ b/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 -f obin-ad-user-creator-compose.yml config` +- `GET /health` diff --git a/obin-ad-user-creator/app/__init__.py b/obin-ad-user-creator/app/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/obin-ad-user-creator/app/main.py b/obin-ad-user-creator/app/main.py new file mode 100644 index 0000000..7d84908 --- /dev/null +++ b/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"""{escape(title)}
{body}
""") + + +@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"
{escape(error)}
" if error else "" + invite_html = "" + if settings.require_invite_code: + invite_html = """ + + + """ + return page(settings.app_title, f""" +

{escape(settings.app_title)}

+

Create a Wheelytho account. After the account is created, you will be sent to setup instructions.

+ {error_html} +
+ + + + + + + {invite_html} + +
+
Password minimum: {settings.minimum_password_length} characters. Your account will be added to the configured AD OU and default access group.
+ """) + + +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""" +

Account created

+

Your account {safe_user} was created{f' for {safe_email}' if safe_email else ''}.

+

Set up 2FA

+ +

Open Wheelytho welcome page

+

Open Authelia directly

+ """) diff --git a/obin-ad-user-creator/requirements.txt b/obin-ad-user-creator/requirements.txt new file mode 100644 index 0000000..4142174 --- /dev/null +++ b/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 diff --git a/obin-ad-user-creator/tests/test_app.py b/obin-ad-user-creator/tests/test_app.py new file mode 100644 index 0000000..3517b0c --- /dev/null +++ b/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