bstein/bstein-dev-home
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

portal: add vaultwarden grandfathered claim

Browse Source
This commit is contained in:
Brad Stein 2026-01-23 22:30:21 -03:00
parent 498e1f4154
commit 4732334c44
5 changed files with 205 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
backend/atlas_portal/keycloak.py
View File

@ -276,6 +276,26 @@ class KeycloakAdminClient:
walk(items)
return sorted(names)
def list_user_groups(self, user_id: str) -> list[str]:
url = (
f"{settings.KEYCLOAK_ADMIN_URL}/admin/realms/{settings.KEYCLOAK_REALM}"
f"/users/{quote(user_id, safe='')}/groups"
)
with httpx.Client(timeout=settings.HTTP_CHECK_TIMEOUT_SEC) as client:
resp = client.get(url, headers=self._headers())
resp.raise_for_status()
items = resp.json()
if not isinstance(items, list):
return []
names: list[str] = []
for item in items:
if not isinstance(item, dict):
continue
name = item.get("name")
if isinstance(name, str) and name:
names.append(name.lstrip("/"))
return names
def add_user_to_group(self, user_id: str, group_id: str) -> None:
url = (
f"{settings.KEYCLOAK_ADMIN_URL}/admin/realms/{settings.KEYCLOAK_REALM}"

119
backend/atlas_portal/routes/access_requests.py
View File

@ -218,7 +218,8 @@ ONBOARDING_STEP_PREREQUISITES: dict[str, set[str]] = {
"jellyfin_tv_setup": {"jellyfin_web_access"},
}
_VAULTWARDEN_READY_STATUSES = {"already_present", "active", "ready"}
VAULTWARDEN_GRANDFATHERED_FLAG = "vaultwarden_grandfathered"
_VAULTWARDEN_READY_STATUSES = {"already_present", "active", "ready", "grandfathered"}
def _normalize_status(status: str) -> str:
@ -241,6 +242,66 @@ def _fetch_completed_onboarding_steps(conn, request_code: str) -> set[str]:
return completed
def _normalize_flag_list(raw: Any) -> set[str]:
if isinstance(raw, list):
return {item for item in raw if isinstance(item, str) and item}
if isinstance(raw, str) and raw:
return {raw}
return set()
def _fetch_request_flags_and_email(conn, request_code: str) -> tuple[set[str], str]:
row = conn.execute(
"SELECT approval_flags, contact_email FROM access_requests WHERE request_code = %s",
(request_code,),
).fetchone()
if not row:
return set(), ""
flags = _normalize_flag_list(row.get("approval_flags"))
email = row.get("contact_email") if isinstance(row, dict) else ""
return flags, (email or "").strip()
def _user_in_group(username: str, group_name: str) -> bool:
if not username or not group_name:
return False
if not admin_client().ready():
return False
try:
user = admin_client().find_user(username) or {}
user_id = user.get("id") if isinstance(user, dict) else None
if not isinstance(user_id, str) or not user_id:
return False
groups = admin_client().list_user_groups(user_id)
except Exception:
return False
return group_name in groups
def _vaultwarden_grandfathered(conn, request_code: str, username: str) -> tuple[bool, str]:
flags, contact_email = _fetch_request_flags_and_email(conn, request_code)
if VAULTWARDEN_GRANDFATHERED_FLAG in flags:
return True, contact_email
if _user_in_group(username, VAULTWARDEN_GRANDFATHERED_FLAG):
return True, contact_email
return False, contact_email
def _resolve_recovery_email(username: str, fallback: str) -> str:
if username and admin_client().ready():
try:
user = admin_client().find_user(username) or {}
user_id = user.get("id") if isinstance(user, dict) else None
if isinstance(user_id, str) and user_id:
full = admin_client().get_user(user_id)
email = full.get("email")
if isinstance(email, str) and email.strip():
return email.strip()
except Exception:
pass
return (fallback or "").strip()
def _password_rotation_requested(conn, request_code: str) -> bool:
row = conn.execute(
"""
@ -438,6 +499,8 @@ def _advance_status(conn, request_code: str, username: str, status: str) -> str:
def _onboarding_payload(conn, request_code: str, username: str) -> dict[str, Any]:
completed_steps = sorted(_completed_onboarding_steps(conn, request_code, username))
password_rotation_requested = _password_rotation_requested(conn, request_code)
grandfathered, contact_email = _vaultwarden_grandfathered(conn, request_code, username)
recovery_email = _resolve_recovery_email(username, contact_email) if grandfathered else ""
return {
"required_steps": list(ONBOARDING_REQUIRED_STEPS),
"optional_steps": sorted(ONBOARDING_OPTIONAL_STEPS),
@ -445,6 +508,10 @@ def _onboarding_payload(conn, request_code: str, username: str) -> dict[str, Any
"keycloak": {
"password_rotation_requested": password_rotation_requested,
},
"vaultwarden": {
"grandfathered": grandfathered,
"recovery_email": recovery_email,
},
}
@ -913,6 +980,7 @@ def register(app) -> None:
code = (payload.get("request_code") or payload.get("code") or "").strip()
step = (payload.get("step") or "").strip()
completed = payload.get("completed")
vaultwarden_claim = bool(payload.get("vaultwarden_claim"))
if not code:
return jsonify({"error": "request_code is required"}), 400
@ -922,6 +990,7 @@ def register(app) -> None:
return jsonify({"error": "step is managed by keycloak"}), 400
username = ""
token_groups: set[str] = set()
bearer = request.headers.get("Authorization", "")
if bearer:
parts = bearer.split(None, 1)
@ -935,11 +1004,14 @@ def register(app) -> None:
except Exception:
return jsonify({"error": "invalid token"}), 401
username = claims.get("preferred_username") or ""
groups = claims.get("groups")
if isinstance(groups, list):
token_groups = {g.lstrip("/") for g in groups if isinstance(g, str) and g}
try:
with connect() as conn:
row = conn.execute(
"SELECT username, status FROM access_requests WHERE request_code = %s",
"SELECT username, status, approval_flags, contact_email FROM access_requests WHERE request_code = %s",
(code,),
).fetchone()
if not row:
@ -956,6 +1028,8 @@ def register(app) -> None:
mark_done = completed
request_username = row.get("username") or ""
approval_flags = _normalize_flag_list(row.get("approval_flags"))
contact_email = (row.get("contact_email") or "").strip()
if mark_done:
prerequisites = ONBOARDING_STEP_PREREQUISITES.get(step, set())
@ -965,6 +1039,17 @@ def register(app) -> None:
if missing:
return jsonify({"error": "step is blocked", "blocked_by": missing}), 409
if step == "vaultwarden_master_password":
if vaultwarden_claim and not username:
return jsonify({"error": "login required"}), 401
grandfathered = (
VAULTWARDEN_GRANDFATHERED_FLAG in approval_flags
or VAULTWARDEN_GRANDFATHERED_FLAG in token_groups
or _user_in_group(request_username, VAULTWARDEN_GRANDFATHERED_FLAG)
)
if vaultwarden_claim and not grandfathered:
return jsonify({"error": "vaultwarden claim not allowed"}), 403
if vaultwarden_claim and not admin_client().ready():
return jsonify({"error": "keycloak admin unavailable"}), 503
try:
_request_keycloak_password_rotation(conn, code, request_username)
except Exception:
@ -972,16 +1057,36 @@ def register(app) -> None:
if request_username and admin_client().ready():
try:
now = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
if vaultwarden_claim:
recovery_email = _resolve_recovery_email(request_username, contact_email)
if not recovery_email:
return jsonify({"error": "recovery email missing"}), 409
admin_client().set_user_attribute(
request_username,
"vaultwarden_email",
recovery_email,
)
admin_client().set_user_attribute(
request_username,
"vaultwarden_status",
"grandfathered",
)
admin_client().set_user_attribute(
request_username,
"vaultwarden_synced_at",
now,
)
else:
admin_client().set_user_attribute(
request_username,
"vaultwarden_status",
"already_present",
)
admin_client().set_user_attribute(
request_username,
"vaultwarden_master_password_set_at",
now,
)
admin_client().set_user_attribute(
request_username,
"vaultwarden_status",
"already_present",
)
except Exception:
return jsonify({"error": "failed to update vaultwarden status"}), 502

17
backend/tests/test_access_requests.py
View File

@ -272,3 +272,20 @@ class AccessRequestTests(TestCase):
data = resp.get_json()
self.assertEqual(resp.status_code, 200)
self.assertEqual(data.get("initial_password"), "temp-pass")
def test_onboarding_payload_includes_vaultwarden_grandfathered(self):
rows = {
"SELECT approval_flags": {
"approval_flags": ["vaultwarden_grandfathered"],
"contact_email": "alice@example.com",
}
}
conn = DummyConn(rows_by_query=rows)
with (
mock.patch.object(ar, "_completed_onboarding_steps", lambda *args, **kwargs: set()),
mock.patch.object(ar, "_password_rotation_requested", lambda *args, **kwargs: False),
):
payload = ar._onboarding_payload(conn, "alice~CODE123", "alice")
vault = payload.get("vaultwarden") or {}
self.assertTrue(vault.get("grandfathered"))
self.assertEqual(vault.get("recovery_email"), "alice@example.com")

4
frontend/src/views/AccountView.vue
View File

@ -489,7 +489,9 @@ const admin = reactive({
selectedFlags: {},
});
const onboardingUrl = ref("/onboarding");
const vaultwardenReady = computed(() => ["ready", "already_present", "active"].includes(vaultwarden.status));
const vaultwardenReady = computed(() =>
["ready", "already_present", "active", "grandfathered"].includes(vaultwarden.status),
);
const vaultwardenDisplayStatus = computed(() => (vaultwardenReady.value ? "ready" : vaultwarden.status));
const vaultwardenOrder = computed(() => (vaultwardenReady.value ? 3 : 0));

56
frontend/src/views/OnboardingView.vue
View File

@ -177,6 +177,31 @@
<p class="muted" v-if="step.description">{{ step.description }}</p>
<div
v-if="step.id === 'vaultwarden_master_password' && vaultwardenGrandfathered"
class="claim-box"
>
<p class="muted">
Already have a Vaultwarden account? Claim it with
<span class="mono">{{ vaultwardenRecoveryEmail || "your recovery email" }}</span>.
This skips the invite flow and keeps your existing vault.
</p>
<button
class="secondary"
type="button"
@click="claimVaultwarden"
:disabled="
loading ||
vaultwardenClaiming ||
!auth.authenticated ||
isStepDone(step.id) ||
isStepBlocked(step.id)
"
>
{{ vaultwardenClaiming ? "Claiming..." : "Claim existing account" }}
</button>
</div>
<ul v-if="step.bullets && step.bullets.length" class="step-bullets">
<li v-for="bullet in step.bullets" :key="bullet">{{ bullet }}</li>
</ul>
@ -318,6 +343,7 @@ const guideShots = ref({});
const guidePage = ref({});
const lightboxShot = ref(null);
const confirmingStepId = ref("");
const vaultwardenClaiming = ref(false);
const showPasswordCard = computed(() => Boolean(initialPassword.value || initialPasswordRevealedAt.value));
const passwordRevealLocked = computed(() => Boolean(!initialPassword.value && initialPasswordRevealedAt.value));
@ -326,6 +352,8 @@ const passwordRevealHint = computed(() =>
? "This password was already revealed and cannot be shown again. Ask an admin to reset it if you missed it."
: "",
);
const vaultwardenGrandfathered = computed(() => Boolean(onboarding.value?.vaultwarden?.grandfathered));
const vaultwardenRecoveryEmail = computed(() => onboarding.value?.vaultwarden?.recovery_email || "");
const STEP_PREREQS = {
vaultwarden_master_password: [],
@ -912,7 +940,7 @@ async function toggleStep(stepId, event) {
await setStepCompletion(stepId, checked);
}
async function setStepCompletion(stepId, completed) {
async function setStepCompletion(stepId, completed, extra = {}) {
if (!requestCode.value.trim()) {
error.value = "Request code is missing.";
return;
@ -927,13 +955,13 @@ async function setStepCompletion(stepId, completed) {
let resp = await requester("/api/access/request/onboarding/attest", {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({ request_code: requestCode.value.trim(), step: stepId, completed }),
body: JSON.stringify({ request_code: requestCode.value.trim(), step: stepId, completed, ...extra }),
});
if ([401, 403].includes(resp.status) && requester === authFetch) {
resp = await fetch("/api/access/request/onboarding/attest", {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({ request_code: requestCode.value.trim(), step: stepId, completed }),
body: JSON.stringify({ request_code: requestCode.value.trim(), step: stepId, completed, ...extra }),
});
}
const data = await resp.json().catch(() => ({}));
@ -987,6 +1015,18 @@ async function confirmStep(step) {
}
}
async function claimVaultwarden() {
if (isStepDone("vaultwarden_master_password") || isStepBlocked("vaultwarden_master_password")) return;
vaultwardenClaiming.value = true;
try {
await setStepCompletion("vaultwarden_master_password", true, { vaultwarden_claim: true });
} catch (err) {
error.value = err?.message || "Failed to claim Vaultwarden account";
} finally {
vaultwardenClaiming.value = false;
}
}
async function runRotationCheck(service) {
if (!auth.authenticated) {
throw new Error("Log in to update onboarding steps.");
@ -1484,6 +1524,16 @@ button.copy:disabled {
text-decoration: underline;
}
.claim-box {
margin-top: 12px;
padding: 10px 12px;
border-radius: 12px;
border: 1px dashed rgba(255, 255, 255, 0.18);
background: rgba(0, 0, 0, 0.2);
display: grid;
gap: 8px;
}
.step-actions {
display: flex;
align-items: center;