bingshuo/hololake-system-architecture
Watch
1
0
Fork
You've already forked hololake-system-architecture
 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity

feat(enterprise): add first-login rotation and receipt bridge

Browse source
This commit is contained in:
冰朔 2026-08-16 20:49:13 +08:00
commit 08f514b9d1
9 changed files with 322 additions and 4 deletions
Download patch file Download diff file Expand all files Collapse all files

91
deployment/enterprise-domain-bootstrap/enterprise-lighthouse/enterprise_identity_service.py
View file

@ -8,6 +8,7 @@ verified against the local Forgejo API and are never stored or logged.
from __future__ import annotations
import base64
import http.cookiejar
import hashlib
import hmac
import json
@ -35,6 +36,9 @@ REGISTRY_PATH = os.environ.get(
FORGEJO_USER_API = os.environ.get(
"GH_ENTERPRISE_FORGEJO_USER_API", "http://127.0.0.1:3341/api/v1/user"
)
FORGEJO_WEB_BASE = os.environ.get(
"GH_ENTERPRISE_FORGEJO_WEB_BASE", "https://guanghu.chat/code"
).rstrip("/")
RECEIPT_KEY = os.environ.get("GH_ENTERPRISE_RECEIPT_KEY", "")
MAX_BODY = 16_384
USERNAME = re.compile(r"^[A-Za-z0-9_-]{1,40}$")
@ -102,6 +106,14 @@ def database() -> sqlite3.Connection:
receipt_hash TEXT NOT NULL,
receipt_signature TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS credential_rotation_receipts (
receipt_id TEXT PRIMARY KEY,
human_number TEXT NOT NULL,
username TEXT NOT NULL,
observed_at INTEGER NOT NULL,
receipt_hash TEXT NOT NULL,
receipt_signature TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS audit (
sequence INTEGER PRIMARY KEY AUTOINCREMENT,
observed_at INTEGER NOT NULL,
@ -152,6 +164,43 @@ def verify_forgejo(username: str, password: str) -> bool:
return False
def rotate_forgejo_password(username: str, current_password: str, new_password: str) -> bool:
"""Use Forgejo's own first-login session to rotate a forced-change password.
This needs no standing admin token: the old credential opens a normal user
session and Forgejo itself admits only the forced password-change form.
"""
jar = http.cookiejar.CookieJar()
opener = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(jar))
try:
opener.open(FORGEJO_WEB_BASE + "/user/login", timeout=15).read()
login = urllib.parse.urlencode(
{"user_name": username, "password": current_password}
).encode()
login_request = urllib.request.Request(
FORGEJO_WEB_BASE + "/user/login", data=login
)
login_request.add_header("Content-Type", "application/x-www-form-urlencoded")
with opener.open(login_request, timeout=15) as response:
response.read()
if not urllib.parse.urlparse(response.geturl()).path.endswith(
"/user/settings/change_password"
):
return False
change = urllib.parse.urlencode(
{"password": new_password, "retype": new_password}
).encode()
change_request = urllib.request.Request(
FORGEJO_WEB_BASE + "/user/settings/change_password", data=change
)
change_request.add_header("Content-Type", "application/x-www-form-urlencoded")
with opener.open(change_request, timeout=15) as response:
response.read()
return verify_forgejo(username, new_password)
except (urllib.error.HTTPError, urllib.error.URLError, TimeoutError, ValueError):
return False
def signed_receipt(payload: dict) -> dict:
if len(RECEIPT_KEY) < 32:
raise RuntimeError("receipt signing key unavailable")
@ -255,6 +304,48 @@ class Handler(BaseHTTPRequestHandler):
payload = self.body()
except (OSError, ValueError, json.JSONDecodeError) as error:
return self.respond(400, {"ok": False, "error": str(error)})
if self.path == "/v1/change-password":
credentials = parse_basic(self.headers.get("Authorization", ""))
human = find_human(registry, str(payload.get("human_number", "")))
new_password = str(payload.get("new_password", ""))
if not credentials or not human:
return self.respond(401, {"ok": False, "error": "enterprise account authentication failed"})
username, current_password = credentials
if not hmac.compare_digest(username, human["username"]):
return self.respond(401, {"ok": False, "error": "enterprise account authentication failed"})
if (
len(new_password) < 14
or len(new_password) > 128
or hmac.compare_digest(current_password, new_password)
or new_password.isdigit()
or new_password.isalpha()
):
return self.respond(400, {"ok": False, "error": "new password does not meet the first-login policy"})
if not rotate_forgejo_password(username, current_password, new_password):
return self.respond(401, {"ok": False, "error": "first-login password rotation failed"})
observed = now()
receipt_id = "GH-CRED-" + uuid.uuid4().hex.upper()
receipt = signed_receipt({
"receipt_id": receipt_id,
"human_number": human["human_number"],
"username": username,
"observed_at": observed,
"password_changed": True,
})
db = database()
try:
db.execute(
"INSERT INTO credential_rotation_receipts VALUES (?,?,?,?,?,?)",
(receipt_id, human["human_number"], username, observed, receipt["receipt_hash"], receipt["receipt_signature"]),
)
db.execute(
"INSERT INTO audit(observed_at,kind,human_number_hash,receipt_id) VALUES (?,?,?,?)",
(observed, "CREDENTIAL_ROTATION", hashlib.sha256(human["human_number"].encode()).hexdigest(), receipt_id),
)
db.commit()
finally:
db.close()
return self.respond(200, {"ok": True, "receipt": receipt})
authenticated = self.authenticated_human(registry, payload)
if not authenticated:
return self.respond(401, {"ok": False, "error": "enterprise account authentication failed"})