coulomb/net-kingdom
2
0
Fork 0
generated from coulomb/repo-seed
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(local-identity): Stage 2 — Keycloak export & bootstrap integration (NK-WP-0002-T02)

Browse Source 
export.py:
  - split_fullname(): last-token strategy (Bernd Worsch → firstName/lastName)
  - _deterministic_id(): uuid5(DNS, "local-identity.{realm}.{username}") for stable,
    re-import-idempotent Keycloak IDs
  - user_to_keycloak(): full Keycloak Admin REST API user representation;
    production_identity mapping applied to username + realm; isolation attributes
    (local_identity_environment, local_identity_generated) always present;
    validate_keycloak_user() called on every conversion to catch schema drift
  - bulk_export_body(): partial import body (ifResourceExists/realm/users)

cli.py: add `export` subcommand
  - export <username>         single user, prints Keycloak JSON
  - export (no args)          bulk; primary users only; stderr note on skipped test users
  - export --include-test     bulk; all users including generated
  - --realm / --if-resource-exists flags

docs/LocalIdentity.md: add two new sections
  - Keycloak import procedure: export → partialImport API → password reset → retire
  - Isolation guarantee: attribute schema, Keycloak Condition authenticator config,
    production_identity mapping walkthrough

tests/test_export.py: 34 new tests (88 total, all passing)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Bernd Worsch
2026-03-02 00:23:39 +01:00
parent 666a56f4ed
commit dad8365e6a
5 changed files with 523 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

91
docs/LocalIdentity.md
View File

@@ -141,6 +141,93 @@ it on a public interface.
| File schema drifting from Keycloak model | `export` command validates against Keycloak representation; schema is versioned |
| Bootstrap store becoming a long-lived crutch | Explicit scope limit: once Keycloak is operational, migrate and stop using Local Identity |
## Keycloak import procedure
Once the Keycloak realm is operational (NK-WP-0001 T06), migrate the primary
user from Local Identity into Keycloak using the partial import endpoint.
**1. Export the primary user:**
```bash
local-identity export --all --realm net-kingdom > /tmp/li-import.json
# By default, only the primary user is exported (test users are excluded).
# Check: the Note line on stderr confirms how many test users were skipped.
```
**2. Import via the Keycloak Admin REST API:**
```bash
# Requires a Keycloak admin token
TOKEN=$(curl -s -X POST https://keycloak.yourdomain.com/realms/master/protocol/openid-connect/token \
-d "client_id=admin-cli&grant_type=password&username=admin&password=<admin-pw>" \
| jq -r .access_token)
curl -s -X POST \
https://keycloak.yourdomain.com/admin/realms/net-kingdom/partialImport \
-H "Authorization: Bearer $TOKEN" \
-H "Content-Type: application/json" \
-d @/tmp/li-import.json
```
**3. Set a password in Keycloak** (Local Identity does not export credentials):
```bash
curl -s -X PUT \
https://keycloak.yourdomain.com/admin/realms/net-kingdom/users/<user-id>/reset-password \
-H "Authorization: Bearer $TOKEN" \
-H "Content-Type: application/json" \
-d '{"type":"password","value":"<new-password>","temporary":false}'
```
**4. Retire Local Identity for this instance:**
Once the user is operational in Keycloak, stop using `local-identity serve`
for this environment and remove the store: `rm -rf ~/.local-identity`.
## Isolation guarantee
All users exported by Local Identity carry the attribute:
```json
"attributes": {
"local_identity_environment": ["local"]
}
```
Generated test users additionally carry `local_identity_generated: ["true"]`.
### Configuring Keycloak to reject local-identity users
Add a condition to the Keycloak browser authentication flow that denies login
for any user with `local_identity_environment = local`. This is a
defence-in-depth measure: even if test users are accidentally imported into a
production realm, they cannot authenticate.
In Keycloak 23+, use a **Conditional Authenticator** with a User Attribute
Condition:
1. In the realm's **Authentication → Flows → browser** flow, add a sub-flow.
2. Add the **Condition - User Attribute** authenticator.
3. Configure: attribute = `local_identity_environment`, value = `local`,
negation = **false** (matches when attribute equals the value).
4. Set the sub-flow to **DENY** when the condition is true.
Alternatively, use a Keycloak script authenticator or a custom policy enforcer.
### Production identity mapping
Before importing, you can assign a `production_identity` block to the user
so the Keycloak username differs from the local username:
```yaml
# ~/.local-identity/users/tegwick.yaml
production_identity:
username: bworsch # the username used in the production realm
realm: net-kingdom
```
Re-run `local-identity export --all` — the exported JSON will use `bworsch`
as the Keycloak username and a deterministic UUID derived from `net-kingdom/bworsch`.
## Relationship to the SSO platform
Local Identity is a complementary workstream to the SSO & MFA Platform
@@ -148,8 +235,4 @@ Local Identity is a complementary workstream to the SSO & MFA Platform
Identity provides the bootstrap path that allows the SSO platform itself to
be set up and tested.
When the Keycloak realm (NK-WP-0001 T06) is operational, primary and test
users can be exported from Local Identity into Keycloak using
`local-identity export` and the Keycloak admin API.
Implementation: see [NK-WP-0002](../workplans/NK-WP-0002-local-identity.md).

60
local-identity/src/local_identity/cli.py
View File

@@ -8,16 +8,21 @@ Commands:
resolved flag > config > system derivation.
list List all users in the store.
show <username> Display a user's YAML record.
export [<username>] Export a single user as Keycloak JSON.
export --all [--realm R] Bulk partial-import body (primary users only).
Add --include-test to include generated users.
Environment:
LOCAL_IDENTITY_HOME Override the store directory (default: ~/.local-identity).
"""
import argparse
import json
import sys
from .gecos import current_username, get_gecos_fullname
from .user import UserRecord, make_test_user
from . import export as export_mod
from . import store
@@ -87,6 +92,38 @@ def cmd_list(args: argparse.Namespace) -> None:
print(f"{u.username:<20} {u.fullname:<30} {u.email:<40} {utype}")
def cmd_export(args: argparse.Namespace) -> None:
if args.username:
# Single-user export
try:
user = store.read_user(args.username)
except FileNotFoundError as exc:
print(str(exc), file=sys.stderr)
sys.exit(1)
kc = export_mod.user_to_keycloak(user, realm=args.realm)
print(json.dumps(kc, indent=2))
else:
# Bulk export
all_users = store.list_users()
if args.include_test:
users = all_users
else:
users = [u for u in all_users if not u.generated]
skipped = len(all_users) - len(users)
if skipped:
print(
f"Note: skipping {skipped} test user(s). "
"Use --include-test to export them.",
file=sys.stderr,
)
body = export_mod.bulk_export_body(
users,
realm=args.realm,
if_resource_exists=args.if_resource_exists,
)
print(json.dumps(body, indent=2))
def cmd_show(args: argparse.Namespace) -> None:
try:
user = store.read_user(args.username)
@@ -129,6 +166,29 @@ def main() -> None:
)
p_init.set_defaults(func=cmd_init)
p_export = sub.add_parser(
"export",
help="Export user(s) as Keycloak-compatible JSON",
)
p_export.add_argument(
"username", nargs="?",
help="Export a single user (omit for bulk export)",
)
p_export.add_argument(
"--realm", default="net-kingdom",
help="Target Keycloak realm name (default: net-kingdom)",
)
p_export.add_argument(
"--include-test", action="store_true",
help="Include generated test users in bulk export",
)
p_export.add_argument(
"--if-resource-exists", default="SKIP",
choices=["SKIP", "OVERWRITE", "FAIL"],
help="Conflict strategy for bulk import (default: SKIP)",
)
p_export.set_defaults(func=cmd_export)
p_list = sub.add_parser("list", help="List all users in the store")
p_list.set_defaults(func=cmd_list)

175
local-identity/src/local_identity/export.py Normal file
View File

@@ -0,0 +1,175 @@
"""
Keycloak export: convert UserRecords to Keycloak Admin REST API representations.
Single-user export:
user_to_keycloak(user, realm="net-kingdom") → dict
Bulk export (Keycloak partial import body):
bulk_export_body(users, realm="net-kingdom") → dict
POST /admin/realms/{realm}/partialImport
Deterministic IDs:
UUIDs are generated with uuid5(DNS, "local-identity.{realm}.{username}") so
repeated exports of the same user produce the same Keycloak ID. This makes
re-imports idempotent when ifResourceExists="SKIP".
Isolation markers:
All exported users carry:
attributes.local_identity_environment = ["local"]
Generated test users additionally carry:
attributes.local_identity_generated = ["true"]
Configure Keycloak to deny authentication for users with
local_identity_environment == "local" as a defence-in-depth measure.
See docs/LocalIdentity.md — "Isolation guarantee" section.
"""
from __future__ import annotations
import uuid
from typing import Sequence
from .user import UserRecord
# ------------------------------------------------------------------ #
# Name splitting #
# ------------------------------------------------------------------ #
def split_fullname(fullname: str) -> tuple[str, str]:
"""
Split a display name into (firstName, lastName).
Strategy: the last whitespace-separated token is lastName; everything
before it is firstName. A single-word name maps to (name, '').
Examples:
"Bernd Worsch" → ("Bernd", "Worsch")
"Jean Claude Damme" → ("Jean Claude", "Damme")
"Madonna" → ("Madonna", "")
"""
parts = fullname.strip().split()
if len(parts) >= 2:
return " ".join(parts[:-1]), parts[-1]
return fullname.strip(), ""
# ------------------------------------------------------------------ #
# Deterministic ID #
# ------------------------------------------------------------------ #
def _deterministic_id(realm: str, username: str) -> str:
"""
Stable UUID5 for a (realm, username) pair.
Using NAMESPACE_DNS keeps it domain-agnostic and collision-resistant.
"""
return str(uuid.uuid5(uuid.NAMESPACE_DNS, f"local-identity.{realm}.{username}"))
# ------------------------------------------------------------------ #
# Schema validation #
# ------------------------------------------------------------------ #
_REQUIRED_FIELDS: dict[str, type] = {
"username": str,
"firstName": str,
"lastName": str,
"email": str,
"enabled": bool,
"emailVerified": bool,
"attributes": dict,
"credentials": list,
}
def validate_keycloak_user(d: dict) -> None:
"""
Minimal structural validation against the expected Keycloak user schema.
Raises ValueError listing every missing or wrongly-typed field, so that
changes to user_to_keycloak() are caught before they reach a live realm.
"""
errors: list[str] = []
for field, expected in _REQUIRED_FIELDS.items():
if field not in d:
errors.append(f"missing required field '{field}'")
elif not isinstance(d[field], expected):
actual = type(d[field]).__name__
errors.append(
f"field '{field}': expected {expected.__name__}, got {actual}"
)
if errors:
raise ValueError(
"Keycloak user schema validation failed:\n"
+ "\n".join(f" - {e}" for e in errors)
)
# ------------------------------------------------------------------ #
# Conversion #
# ------------------------------------------------------------------ #
def user_to_keycloak(user: UserRecord, realm: str = "net-kingdom") -> dict:
"""
Convert a UserRecord to a Keycloak Admin REST API user representation.
production_identity mapping:
If the user carries a production_identity block, the Keycloak username
and realm are taken from it. This allows a local username of 'tegwick'
to map to a different production username/realm before import.
"""
first_name, last_name = split_fullname(user.fullname)
kc_username = user.username
kc_realm = realm
if user.production_identity:
kc_username = user.production_identity.username
if user.production_identity.realm:
kc_realm = user.production_identity.realm
attrs: dict[str, list[str]] = {
"local_identity_environment": [user.environment],
}
if user.generated:
attrs["local_identity_generated"] = ["true"]
d: dict = {
"id": _deterministic_id(kc_realm, kc_username),
"username": kc_username,
"firstName": first_name,
"lastName": last_name,
"email": user.email,
"enabled": True,
"emailVerified": False,
"attributes": attrs,
"credentials": [],
"requiredActions": [],
"groups": [],
"realmRoles": [],
"clientRoles": {},
}
validate_keycloak_user(d)
return d
# ------------------------------------------------------------------ #
# Bulk export #
# ------------------------------------------------------------------ #
def bulk_export_body(
users: Sequence[UserRecord],
realm: str = "net-kingdom",
if_resource_exists: str = "SKIP",
) -> dict:
"""
Produce a Keycloak partial import request body.
POST /admin/realms/{realm}/partialImport
Content-Type: application/json
ifResourceExists values: SKIP | OVERWRITE | FAIL
SKIP is the safe default — existing users are left unchanged.
"""
return {
"ifResourceExists": if_resource_exists,
"realm": realm,
"users": [user_to_keycloak(u, realm=realm) for u in users],
}

198
local-identity/tests/test_export.py Normal file
View File

@@ -0,0 +1,198 @@
"""Tests for the Keycloak export module."""
import json
import uuid
import pytest
from local_identity.export import (
_deterministic_id,
bulk_export_body,
split_fullname,
user_to_keycloak,
validate_keycloak_user,
)
from local_identity.user import ProductionIdentity, UserRecord, make_test_user
PRIMARY = UserRecord(
username="tegwick",
fullname="Bernd Worsch",
email="bernd.worsch@gmail.com",
)
# ------------------------------------------------------------------ #
# split_fullname #
# ------------------------------------------------------------------ #
class TestSplitFullname:
def test_two_words(self):
assert split_fullname("Bernd Worsch") == ("Bernd", "Worsch")
def test_three_words(self):
assert split_fullname("Jean Claude Damme") == ("Jean Claude", "Damme")
def test_single_word(self):
assert split_fullname("Madonna") == ("Madonna", "")
def test_extra_whitespace_stripped(self):
assert split_fullname(" Bernd Worsch ") == ("Bernd", "Worsch")
def test_non_ascii(self):
first, last = split_fullname("Ärger Müller")
assert first == "Ärger"
assert last == "Müller"
# ------------------------------------------------------------------ #
# _deterministic_id #
# ------------------------------------------------------------------ #
class TestDeterministicId:
def test_stable_across_calls(self):
assert _deterministic_id("net-kingdom", "tegwick") == _deterministic_id("net-kingdom", "tegwick")
def test_different_realm_gives_different_id(self):
assert _deterministic_id("realm-a", "u") != _deterministic_id("realm-b", "u")
def test_different_username_gives_different_id(self):
assert _deterministic_id("r", "alice") != _deterministic_id("r", "bob")
def test_is_valid_uuid(self):
uid = _deterministic_id("net-kingdom", "tegwick")
uuid.UUID(uid) # must not raise
# ------------------------------------------------------------------ #
# user_to_keycloak #
# ------------------------------------------------------------------ #
class TestUserToKeycloak:
def test_username(self):
assert user_to_keycloak(PRIMARY)["username"] == "tegwick"
def test_name_split(self):
kc = user_to_keycloak(PRIMARY)
assert kc["firstName"] == "Bernd"
assert kc["lastName"] == "Worsch"
def test_email(self):
assert user_to_keycloak(PRIMARY)["email"] == "bernd.worsch@gmail.com"
def test_enabled_and_email_verified(self):
kc = user_to_keycloak(PRIMARY)
assert kc["enabled"] is True
assert kc["emailVerified"] is False
def test_environment_attribute(self):
kc = user_to_keycloak(PRIMARY)
assert kc["attributes"]["local_identity_environment"] == ["local"]
def test_primary_user_has_no_generated_attribute(self):
kc = user_to_keycloak(PRIMARY)
assert "local_identity_generated" not in kc["attributes"]
def test_test_user_has_generated_attribute(self):
t = make_test_user(PRIMARY, 1)
kc = user_to_keycloak(t)
assert kc["attributes"]["local_identity_generated"] == ["true"]
def test_structural_fields_present(self):
kc = user_to_keycloak(PRIMARY)
for field in ("credentials", "requiredActions", "groups", "realmRoles", "clientRoles"):
assert field in kc, f"missing field '{field}'"
def test_id_is_deterministic(self):
assert user_to_keycloak(PRIMARY)["id"] == user_to_keycloak(PRIMARY)["id"]
def test_id_is_valid_uuid(self):
uuid.UUID(user_to_keycloak(PRIMARY)["id"])
def test_json_serializable(self):
json.dumps(user_to_keycloak(PRIMARY)) # must not raise
def test_production_identity_overrides_username(self):
u = UserRecord(
username="tegwick",
fullname="Bernd Worsch",
email="bernd.worsch@gmail.com",
production_identity=ProductionIdentity(username="bworsch", realm="prod"),
)
assert user_to_keycloak(u)["username"] == "bworsch"
def test_production_identity_realm_used_in_id(self):
u = UserRecord(
username="tegwick",
fullname="Bernd Worsch",
email="bernd.worsch@gmail.com",
production_identity=ProductionIdentity(username="bworsch", realm="prod"),
)
expected_id = _deterministic_id("prod", "bworsch")
assert user_to_keycloak(u)["id"] == expected_id
def test_default_realm_used_when_no_production_identity(self):
expected_id = _deterministic_id("my-realm", "tegwick")
assert user_to_keycloak(PRIMARY, realm="my-realm")["id"] == expected_id
# ------------------------------------------------------------------ #
# validate_keycloak_user #
# ------------------------------------------------------------------ #
class TestValidateKeycloakUser:
def test_valid_passes(self):
validate_keycloak_user(user_to_keycloak(PRIMARY)) # must not raise
def test_missing_field_raises(self):
d = user_to_keycloak(PRIMARY)
del d["username"]
with pytest.raises(ValueError, match="username"):
validate_keycloak_user(d)
def test_wrong_type_raises(self):
d = user_to_keycloak(PRIMARY)
d["enabled"] = "yes"
with pytest.raises(ValueError, match="enabled"):
validate_keycloak_user(d)
def test_multiple_errors_all_reported(self):
d = user_to_keycloak(PRIMARY)
del d["username"]
del d["email"]
with pytest.raises(ValueError) as exc_info:
validate_keycloak_user(d)
msg = str(exc_info.value)
assert "username" in msg
assert "email" in msg
# ------------------------------------------------------------------ #
# bulk_export_body #
# ------------------------------------------------------------------ #
class TestBulkExportBody:
def test_required_keys_present(self):
body = bulk_export_body([PRIMARY])
assert "ifResourceExists" in body
assert "realm" in body
assert "users" in body
def test_default_if_resource_exists(self):
assert bulk_export_body([PRIMARY])["ifResourceExists"] == "SKIP"
def test_custom_realm(self):
assert bulk_export_body([PRIMARY], realm="my-realm")["realm"] == "my-realm"
def test_user_count(self):
users = [PRIMARY, make_test_user(PRIMARY, 1), make_test_user(PRIMARY, 2)]
assert len(bulk_export_body(users)["users"]) == 3
def test_empty_user_list(self):
assert bulk_export_body([])["users"] == []
def test_overwrite_strategy(self):
body = bulk_export_body([PRIMARY], if_resource_exists="OVERWRITE")
assert body["ifResourceExists"] == "OVERWRITE"
def test_json_serializable(self):
json.dumps(bulk_export_body([PRIMARY, make_test_user(PRIMARY, 1)]))

5
workplans/NK-WP-0002-local-identity.md
View File

@@ -108,8 +108,9 @@ unit tests passing.
```task
id: NK-WP-0002-T02
state_hub_task_id: 5ea6e68d-7ebe-4ea7-b92e-61aac17ff04c
status: todo
status: done
priority: high
commit: c TBD
```
Extend user schema with optional `production_identity` block (`username`,
@@ -223,7 +224,7 @@ expiry and revocation functional.
## Deliverables Checklist
- [x] `~/.local-identity/` store initialised from Linux identity; test users generated
- [ ] `local-identity list / show / export` working; Keycloak export validated
- [x] `local-identity list / show / export` working; Keycloak export validated
- [ ] Minimal OIDC server passes conformance smoke test; binds localhost only
- [ ] Filesystem permissions enforced on startup; `security-check` passes
- [ ] Audit log recording all auth events