bob@example.com" in page.text
+ assert "cannot be stored" not in page.text
+ assert cfg.local_users[0].keys == []
+ issued = client.post("/user/keys", json={"id": "laptop", "readonly": True}, headers=headers)
+ assert issued.status_code == 200, issued.text
+ plaintext = issued.json()["secret"]
+ assert len(cfg.local_users) == 2
+ created = cfg.local_users[1]
+ assert created.user == "bob@example.com"
+ assert created.email == "bob@example.com"
+ assert created.password is None
+ assert created.pass_hash == ""
+ assert created.assume_anonymous is False
+ assert [key.id for key in cfg.local_users[0].keys] == []
+ assert created.keys[0].readonly is True
+ assert verify_password(plaintext, created.keys[0].secret, config_salt=_SALT)
+ on_disk = json.loads(path.read_text(encoding="utf-8"))
+ stored = on_disk["local_users"][1]
+ assert "pass" not in stored
+ assert stored["pass_hash"] == ""
+ assert plaintext not in path.read_text(encoding="utf-8")
+ assert authenticate_local_user("bob@example.com", plaintext) is None
+ matched = authenticate_http_key("bob@example.com", plaintext)
+ assert matched is not None
+ assert matched[0].email == "bob@example.com"
+ assert matched[1] is True
+ assert authenticate_http_key("BOB@example.com", plaintext) == matched
+ assert authenticate_http_key("admin", plaintext) is None
+ again = client.post("/user/keys", json={"id": "phone"}, headers=headers)
+ assert again.status_code == 200, again.text
+ assert len(cfg.local_users) == 2
+ assert [key.id for key in created.keys] == ["laptop", "phone"]
+ revoked = client.delete("/user/keys", params={"id": "laptop"}, headers=headers)
+ assert revoked.status_code == 204
+ assert [key.id for key in created.keys] == ["phone"]
+ no_email = create_session("admin", auth_method="oauth")
+ missing = client.post(
+ "/user/keys",
+ json={"id": "laptop"},
+ headers={"Authorization": f"Bearer {no_email}"},
+ )
+ assert missing.status_code == 404
+ assert missing.json()["detail"] == "An email is required to store API keys"
+ blocked = client.get("/user", headers={"Authorization": f"Bearer {no_email}"})
+ assert "cannot be stored" in blocked.text
+
+
+def test_oauth_key_name_collision_does_not_attach_to_the_other_account(tmp_path: Path) -> None:
+ cfg = AuthConfig(
+ auth_mode="local",
+ salt=_SALT,
+ local_users=[LocalUser(user="taken@example.com", password="secret", email="other@example.com")],
+ )
+ path = tmp_path / "auth.json"
+ client = TestClient(app)
+ with (
+ patch.object(settings, "AUTH", True),
+ patch.object(settings, "AUTH_CONFIG", str(path)),
+ patch("pygitweb.auth_config.auth_config", cfg),
+ patch("pygitweb.auth.auth_config", cfg),
+ patch("pygitweb.main.auth_config", cfg),
+ ):
+ token = create_session("taken@example.com", email="taken@example.com", auth_method="oauth")
+ response = client.post(
+ "/user/keys",
+ json={"id": "laptop"},
+ headers={"Authorization": f"Bearer {token}"},
+ )
+ assert response.status_code == 409
+ assert response.json()["detail"] == "A local account already uses this name"
+ assert len(cfg.local_users) == 1
+ assert cfg.local_users[0].keys == []
+
+
+def test_oauth_key_user_is_removed_when_save_fails(tmp_path: Path) -> None:
+ cfg = _config()
+ path = tmp_path / "auth.json"
+ client = TestClient(app, raise_server_exceptions=False)
+ with (
+ patch.object(settings, "AUTH", True),
+ patch.object(settings, "AUTH_CONFIG", str(path)),
+ patch("pygitweb.auth_config.auth_config", cfg),
+ patch("pygitweb.auth.auth_config", cfg),
+ patch("pygitweb.main.auth_config", cfg),
+ patch("pygitweb.auth.write_auth_config", side_effect=OSError("disk")),
+ ):
+ token = create_session("admin", email="bob@example.com", auth_method="oauth")
+ response = client.post(
+ "/user/keys",
+ json={"id": "laptop"},
+ headers={"Authorization": f"Bearer {token}"},
+ )
+ assert response.status_code == 500
+ assert [user.user for user in cfg.local_users] == ["admin"]
+ assert cfg.local_users[0].keys == []
diff --git a/pygitweb/src/pygitweb/auth.py b/pygitweb/src/pygitweb/auth.py
index c44a050..940f0da 100644
--- a/pygitweb/src/pygitweb/auth.py
+++ b/pygitweb/src/pygitweb/auth.py
@@ -49,7 +49,7 @@ from pygitweb.permissions import (
ref_write_grant,
)
from pygitweb.projects import git_get_project_owner
-from pygitweb.sessions import create_session, get_session, revoke_session
+from pygitweb.sessions import AuthMethod, create_session, get_session, revoke_session
from pygitweb.templates_env import render_page
ACCESS_TOKEN_COOKIE_NAME = "pygitweb_access_token"
@@ -683,19 +683,53 @@ class IssuedApiKey(BaseModel):
time_expiry: datetime | None = None
-def _local_user_for_principal(username: str, email: str | None) -> LocalUser | None:
- for entry in auth_config.local_users:
- if _const_eq(entry.user, username):
- return entry
+def _local_user_by_email(email: str | None) -> LocalUser | None:
if not email:
return None
- folded = email.casefold()
+ folded = email.strip().casefold()
+ if not folded:
+ return None
for entry in auth_config.local_users:
if entry.email and _const_eq(entry.email.casefold(), folded):
return entry
return None
+def _local_user_for_principal(username: str, email: str | None) -> LocalUser | None:
+ for entry in auth_config.local_users:
+ if _const_eq(entry.user, username):
+ return entry
+ return _local_user_by_email(email)
+
+
+def _key_owner(username: str, email: str | None, auth_method: AuthMethod) -> LocalUser | None:
+ if auth_method == "oauth":
+ return _local_user_by_email(email)
+ return _local_user_for_principal(username, email)
+
+
+def _new_passwordless_user(email: str | None, auth_method: AuthMethod) -> LocalUser:
+ if auth_method != "oauth":
+ raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="No local account for API keys")
+ stored = email.strip() if email else ""
+ if len(stored) < 2:
+ raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="An email is required to store API keys")
+ for entry in auth_config.local_users:
+ if _const_eq(entry.user.casefold(), stored.casefold()):
+ raise HTTPException(
+ status_code=status.HTTP_409_CONFLICT,
+ detail="A local account already uses this name",
+ )
+ return LocalUser(user=stored, email=stored)
+
+
+def _drop_local_user(user: LocalUser) -> None:
+ for index, entry in enumerate(auth_config.local_users):
+ if entry is user:
+ del auth_config.local_users[index]
+ return
+
+
def _key_expiry(issued: datetime) -> datetime | None:
days = settings.API_KEY_EXPIRY_DAYS
if days <= 0:
@@ -719,11 +753,20 @@ def _public_key_rows(user: LocalUser) -> list[dict[str, str | bool | None]]:
return rows
-def issue_api_key(username: str, email: str | None, *, key_id: str, readonly: bool) -> IssuedApiKey:
+def issue_api_key(
+ username: str,
+ email: str | None,
+ *,
+ key_id: str,
+ readonly: bool,
+ auth_method: AuthMethod = "local",
+) -> IssuedApiKey:
with _API_KEY_LOCK:
- user = _local_user_for_principal(username, email)
+ user = _key_owner(username, email, auth_method)
+ created = False
if user is None:
- raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="No local account for API keys")
+ user = _new_passwordless_user(email, auth_method)
+ created = True
if len(user.keys) >= settings.API_KEY_MAX_COUNT:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
@@ -740,12 +783,17 @@ def issue_api_key(username: str, email: str | None, *, key_id: str, readonly: bo
time_issued=issued,
time_expiry=_key_expiry(issued),
)
+ if created:
+ auth_config.local_users.append(user)
previous = list(user.keys)
user.keys = [*previous, stored]
try:
_persist_auth_config()
except Exception:
- user.keys = previous
+ if created:
+ _drop_local_user(user)
+ else:
+ user.keys = previous
raise
return IssuedApiKey(
id=stored.id,
@@ -756,9 +804,15 @@ def issue_api_key(username: str, email: str | None, *, key_id: str, readonly: bo
)
-def revoke_api_key(username: str, email: str | None, key_id: str) -> None:
+def revoke_api_key(
+ username: str,
+ email: str | None,
+ key_id: str,
+ *,
+ auth_method: AuthMethod = "local",
+) -> None:
with _API_KEY_LOCK:
- user = _local_user_for_principal(username, email)
+ user = _key_owner(username, email, auth_method)
if user is None:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="No local account for API keys")
remaining = [key for key in user.keys if not _const_eq(key.id, key_id)]
@@ -773,7 +827,26 @@ def revoke_api_key(username: str, email: str | None, key_id: str) -> None:
raise
-def _session_identity(token: str | None) -> tuple[str, str | None]:
+def _account_keys(
+ username: str,
+ email: str | None,
+ auth_method: AuthMethod,
+) -> tuple[list[dict[str, str | bool | None]] | None, str | None]:
+ if auth_method == "oauth":
+ user = _local_user_by_email(email)
+ if user is not None:
+ return _public_key_rows(user), user.user
+ stored = email.strip() if email else ""
+ if len(stored) < 2:
+ return None, None
+ return [], stored
+ user = _local_user_for_principal(username, email)
+ if user is None:
+ return None, None
+ return _public_key_rows(user), user.user
+
+
+def _session_identity(token: str | None) -> tuple[str, str | None, AuthMethod]:
ensure_active_user_if_auth_enabled(token)
rec = get_session(token or "")
if rec is None:
@@ -782,7 +855,7 @@ def _session_identity(token: str | None) -> tuple[str, str | None]:
detail="Not authenticated",
headers={"WWW-Authenticate": "Bearer"},
)
- return rec.username, rec.email
+ return rec.username, rec.email, rec.auth_method
@auth_router.get(
@@ -808,17 +881,18 @@ async def user_account_page(
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
detail="Authentication is misconfigured",
)
- u = decode_access_token(token or "") if token else None
- if u is None:
- return RedirectResponse(url="/login?next=/user", status_code=303)
rec = get_session(token or "")
- email = rec.email if rec is not None else None
- local_user = _local_user_for_principal(u.username, email)
+ if rec is None:
+ return RedirectResponse(url="/login?next=/user", status_code=303)
+ api_keys, api_key_user = _account_keys(rec.username, rec.email, rec.auth_method)
+ if api_key_user is not None and _const_eq(api_key_user, rec.username):
+ api_key_user = None
return render_page(
"user_profile.html",
title=f"{settings.SITE_NAME} - Account",
- username=u.username,
- api_keys=_public_key_rows(local_user) if local_user is not None else None,
+ username=rec.username,
+ api_keys=api_keys,
+ api_key_user=api_key_user,
api_key_max=settings.API_KEY_MAX_COUNT,
api_key_expiry_days=settings.API_KEY_EXPIRY_DAYS,
)
@@ -830,8 +904,10 @@ async def user_account_page(
tags=[TAG_AUTH],
summary="Issue an API key",
description=(
- "Creates a Smart HTTP API key for the signed-in local user. "
- "The plaintext secret is returned once. Only its scrypt hash is stored."
+ "Creates a Smart HTTP API key for the signed-in user. "
+ "An OAuth session is stored as a passwordless local account, matched by email, "
+ "the first time it issues a key. The plaintext secret is returned once. "
+ "Only its scrypt hash is stored."
),
)
async def issue_user_api_key(
@@ -840,8 +916,8 @@ async def issue_user_api_key(
) -> IssuedApiKey:
if not settings.AUTH:
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Authentication is disabled")
- username, email = _session_identity(token)
- return issue_api_key(username, email, key_id=body.id, readonly=body.readonly)
+ username, email, auth_method = _session_identity(token)
+ return issue_api_key(username, email, key_id=body.id, readonly=body.readonly, auth_method=auth_method)
@auth_router.delete(
@@ -849,7 +925,7 @@ async def issue_user_api_key(
status_code=status.HTTP_204_NO_CONTENT,
tags=[TAG_AUTH],
summary="Revoke an API key",
- description="Removes one Smart HTTP API key from the signed-in local user.",
+ description="Removes one Smart HTTP API key from the signed-in user.",
)
async def revoke_user_api_key(
token: Annotated[str | None, Depends(access_token_from_request)],
@@ -857,8 +933,8 @@ async def revoke_user_api_key(
) -> Response:
if not settings.AUTH:
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Authentication is disabled")
- username, email = _session_identity(token)
- revoke_api_key(username, email, id.strip())
+ username, email, auth_method = _session_identity(token)
+ revoke_api_key(username, email, id.strip(), auth_method=auth_method)
return Response(status_code=status.HTTP_204_NO_CONTENT)
diff --git a/pygitweb/src/pygitweb/templates/user_profile.html b/pygitweb/src/pygitweb/templates/user_profile.html
index 7031a82..bacd61d 100644
--- a/pygitweb/src/pygitweb/templates/user_profile.html
+++ b/pygitweb/src/pygitweb/templates/user_profile.html
@@ -22,6 +22,9 @@
{% if api_keys is none %}
No local account matches this session, so API keys cannot be stored.
{% else %} + {% if api_key_user %} +Smart HTTP username for these keys: {{ api_key_user | e }}
Copy this key now. It will not be shown again.