Dw/devtests password credential

DEVELOPMENT.md

@@ -21,7 +21,7 @@ service: `docker compose --profile dev restart api-dev` (or `app-dev`).
 
 ## Logging in
 
-The web app needs a logged-in session. Set up GitHub OAuth once:
+The web app needs a logged-in session. For a realistic login flow, set up GitHub OAuth once:
 
 1. Create an OAuth App at https://github.com/settings/developers
    - Homepage URL: `http://localhost:8081`
@@ -32,6 +32,20 @@ The web app needs a logged-in session. Set up GitHub OAuth once:
    `python3 -c 'import secrets; print(secrets.token_hex(64))'`).
 4. **Login via GitHub** in the web app.
 
+For local development and testing, you can instead create a testing-only
+username/password user. This only works when `PYDATALAB_TESTING=true`; the route and
+feature flag are disabled otherwise.
+
+```bash
+docker compose exec api-dev /opt/.venv/bin/invoke dev.create-test-user \
+  --username alice \
+  --password alice \
+  --display-name "Alice"
+```
+
+Then open the web app, click **Login/Register**, and use the testing
+username/password login.
+
 
 ## Handy commands
 

pydatalab/src/pydatalab/feature_flags.py

@@ -16,6 +16,8 @@ class AuthMechanisms(BaseModel):
     email: bool = False
     google: bool = False
     microsoft: bool = False
+    # Testing-only username/password login, never intended as a production auth mechanism.
+    testing_username_password: bool = False
 
 
 class AIIntegrations(BaseModel):
@@ -58,6 +60,7 @@ def check_feature_flags(app):
     object reported by the API for use in UIs.
 
     """
+    FEATURE_FLAGS.auth_mechanisms.testing_username_password = bool(CONFIG.TESTING)
 
     if CONFIG.EMAIL_AUTH_SMTP_SETTINGS is None:
         LOGGER.warning(

pydatalab/src/pydatalab/main.py

@@ -220,7 +220,7 @@ def provider_label(identity_type) -> str:
 def register_endpoints(app: Flask):
     """Loops through the implemented endpoints, blueprints and error handlers adds them to the app."""
     from pydatalab.errors import ERROR_HANDLERS
-    from pydatalab.routes import BLUEPRINTS, OAUTH, __api_version__
+    from pydatalab.routes import BLUEPRINTS, __api_version__, get_login_blueprints
 
     major, minor, patch = __api_version__.split(".")
     versions = ["", f"v{major}", f"v{major}.{minor}", f"v{major}.{minor}.{patch}"]
@@ -231,8 +231,8 @@ def register_endpoints(app: Flask):
                 bp, url_prefix=f"{CONFIG.ROOT_PATH}{ver}", name=f"{ver}/{bp.name}"
             )
 
-    for bp in OAUTH:  # type: ignore
-        app.register_blueprint(OAUTH[bp], url_prefix=f"{CONFIG.ROOT_PATH}login")  # type: ignore
+    for bp in get_login_blueprints():
+        app.register_blueprint(bp, url_prefix=f"{CONFIG.ROOT_PATH}login")
 
     for exception_type, handler in ERROR_HANDLERS:
         app.register_error_handler(exception_type, handler)

pydatalab/src/pydatalab/routes/__init__.py

@@ -1,3 +1,16 @@
-from pydatalab.routes.v0_1 import BLUEPRINTS, OAUTH, OAUTH_PROXIES, __api_version__
+from pydatalab.routes.v0_1 import (
+    BLUEPRINTS,
+    OAUTH,
+    OAUTH_PROXIES,
+    __api_version__,
+    get_login_blueprints,
+)
 
-__all__ = ("ENDPOINTS", "__api_version__", "OAUTH", "OAUTH_PROXIES", "BLUEPRINTS")
+__all__ = (
+    "ENDPOINTS",
+    "__api_version__",
+    "OAUTH",
+    "OAUTH_PROXIES",
+    "BLUEPRINTS",
+    "get_login_blueprints",
+)

pydatalab/src/pydatalab/routes/v0_1/__init__.py

@@ -2,7 +2,7 @@
 
 from ._version import __api_version__
 from .admin import ADMIN
-from .auth import AUTH, OAUTH, OAUTH_PROXIES
+from .auth import AUTH, OAUTH, OAUTH_PROXIES, get_login_blueprints
 from .blocks import BLOCKS
 from .collections import COLLECTIONS
 from .export import EXPORT
@@ -31,4 +31,10 @@
     EXPORT,
 )
 
-__all__ = ("BLUEPRINTS", "OAUTH", "__api_version__", "OAUTH_PROXIES")
+__all__ = (
+    "BLUEPRINTS",
+    "OAUTH",
+    "__api_version__",
+    "OAUTH_PROXIES",
+    "get_login_blueprints",
+)

pydatalab/src/pydatalab/routes/v0_1/auth.py

@@ -18,7 +18,7 @@
 from flask_dance.consumer import OAuth2ConsumerBlueprint, oauth_authorized
 from flask_login import current_user, login_user
 from flask_login.utils import LocalProxy
-from werkzeug.exceptions import BadRequest, Forbidden
+from werkzeug.exceptions import BadRequest, Forbidden, Unauthorized
 
 from pydatalab.config import CONFIG
 from pydatalab.errors import UserRegistrationForbidden
@@ -28,6 +28,7 @@
 from pydatalab.models.people import AccountStatus, Identity, IdentityType, Person
 from pydatalab.mongo import flask_mongo, insert_pydantic_model_fork_safe
 from pydatalab.send_email import send_mail
+from pydatalab.testing_username_password_auth import verify_testing_username_password_credential
 
 KEY_LENGTH: int = 32
 LINK_EXPIRATION: datetime.timedelta = datetime.timedelta(hours=1)
@@ -402,6 +403,8 @@ def wrapped_login_user(*args, **kwargs):
 
 EMAIL_BLUEPRINT = Blueprint("email", __name__)
 
+TESTING_USERNAME_PASSWORD_BLUEPRINT = Blueprint("testing_username_password", __name__)
+
 AUTH = Blueprint("auth", __name__)
 
 
@@ -426,6 +429,15 @@ def wrapped_login_user(*args, **kwargs):
 }
 """A dictionary of Flask blueprints corresponding to the supported OAuth providers."""
 
+
+def get_login_blueprints() -> tuple[Blueprint, ...]:
+    """Return blueprints registered under /login."""
+    login_blueprints = tuple(OAUTH.values())
+    if CONFIG.TESTING:
+        login_blueprints += (TESTING_USERNAME_PASSWORD_BLUEPRINT,)
+    return login_blueprints
+
+
 OAUTH_PROXIES: dict[IdentityType, LocalProxy] = {
     IdentityType.ORCID: orcid,
     IdentityType.GITHUB: github,
@@ -862,6 +874,34 @@ def email_logged_in():
     return redirect(referer, 307)
 
 
+@TESTING_USERNAME_PASSWORD_BLUEPRINT.route("/testing-username-password", methods=["POST"])
+def testing_username_password_login():
+    """Testing-only username/password login."""
+    if not CONFIG.TESTING:
+        raise Forbidden("Username/password login is only available in testing mode.")
+
+    request_json = request.get_json() or {}
+    username = request_json.get("username")
+    password = request_json.get("password")
+    if not username or not password:
+        raise Unauthorized("Invalid username or password.")
+
+    user_id = verify_testing_username_password_credential(username, password)
+    if not user_id:
+        raise Unauthorized("Invalid username or password.")
+
+    try:
+        user_model = get_by_id(str(user_id))
+    except (StopIteration, ValueError):
+        user_model = None
+
+    if user_model is None or user_model.account_status == AccountStatus.DEACTIVATED:
+        raise Unauthorized("Invalid username or password.")
+
+    wrapped_login_user(user_model)
+    return jsonify({"status": "success"}), 200
+
+
 @oauth_authorized.connect_via(OAUTH[IdentityType.GITHUB])
 def github_logged_in(blueprint, token):
     """This Flask signal hooks into any attempt to use the GitHub blueprint, and will

pydatalab/src/pydatalab/testing_username_password_auth.py

@@ -0,0 +1,67 @@
+"""Small file-backed credentials for testing-only username/password login."""
+
+from __future__ import annotations
+
+import json
+import os
+from pathlib import Path
+
+from pydantic import ValidationError
+from werkzeug.security import check_password_hash, generate_password_hash
+
+from pydatalab.config import CONFIG
+from pydatalab.models.utils import HumanReadableIdentifier
+
+TESTING_USERNAME_PASSWORD_AUTH_FILENAME = ".testing_username_password_credentials.json"  # noqa: S105
+
+
+def testing_username_password_credentials_path() -> Path:
+    return Path(CONFIG.FILE_DIRECTORY) / TESTING_USERNAME_PASSWORD_AUTH_FILENAME
+
+
+def load_testing_username_password_credentials() -> dict:
+    path = testing_username_password_credentials_path()
+    if not path.is_file():
+        return {}
+
+    try:
+        data = json.loads(path.read_text())
+    except json.JSONDecodeError:
+        return {}
+
+    return data if isinstance(data, dict) else {}
+
+
+def save_testing_username_password_credentials(credentials: dict) -> None:
+    path = testing_username_password_credentials_path()
+    path.parent.mkdir(parents=True, exist_ok=True)
+    tmp_path = path.with_suffix(path.suffix + ".tmp")
+    tmp_path.write_text(json.dumps(credentials, indent=2, sort_keys=True) + "\n")
+    os.replace(tmp_path, path)
+
+
+def set_testing_username_password_credential(username: str, user_id: str, password: str) -> None:
+    username = str(HumanReadableIdentifier(username))
+    credentials = load_testing_username_password_credentials()
+    credentials[username] = {
+        "user_id": str(user_id),
+        "password_hash": generate_password_hash(password),
+    }
+    save_testing_username_password_credentials(credentials)
+
+
+def verify_testing_username_password_credential(username: str, password: str) -> str | None:
+    try:
+        username = str(HumanReadableIdentifier(username))
+    except (ValueError, ValidationError):
+        return None
+
+    credential = load_testing_username_password_credentials().get(username)
+    if not credential:
+        return None
+
+    password_hash = credential.get("password_hash")
+    if not password_hash or not check_password_hash(password_hash, password):
+        return None
+
+    return credential.get("user_id")

pydatalab/tasks.py

@@ -165,6 +165,90 @@ def serve(_, host: str = "127.0.0.1", port: int = 5001, reload: bool = True, tes
 dev.add_task(serve)
 
 
+@task(
+    aliases=["create-test-user"],
+    help={
+        "username": "Testing username/password username",
+        "password": "Testing username/password password",
+        "display_name": "Display name for a newly-created or updated user",
+        "contact_email": "Optional contact email for the user",
+        "role": "User role: user, manager, or admin",
+        "account_status": "Account status: active, unverified, or deactivated",
+    },
+)
+def create_testing_username_password_user(
+    _,
+    username: str,
+    password: str,
+    display_name: str | None = None,
+    contact_email: str | None = None,
+    role: str = "user",
+    account_status: str = "active",
+):
+    """Create or update a testing-only username/password user."""
+    from bson import ObjectId
+    from pydantic import ValidationError
+
+    from pydatalab.config import CONFIG
+    from pydatalab.models.people import AccountStatus, Person
+    from pydatalab.models.utils import HumanReadableIdentifier, UserRole
+    from pydatalab.mongo import get_database
+    from pydatalab.testing_username_password_auth import (
+        load_testing_username_password_credentials,
+        set_testing_username_password_credential,
+    )
+
+    if not CONFIG.TESTING:
+        raise SystemExit(
+            "Testing username/password users are only available when CONFIG.TESTING is true. "
+            "Set PYDATALAB_TESTING=1 before running this task."
+        )
+
+    try:
+        username = str(HumanReadableIdentifier(username))
+    except ValidationError as exc:
+        raise SystemExit(f"Invalid username {username!r}: {exc}") from None
+    try:
+        role_value = UserRole(role)
+    except ValueError:
+        raise SystemExit(f"Invalid role {role!r}. Must be one of {[r.value for r in UserRole]}.")
+
+    try:
+        status_value = AccountStatus(account_status)
+    except ValueError:
+        raise SystemExit(
+            f"Invalid account status {account_status!r}. "
+            f"Must be one of {[s.value for s in AccountStatus]}."
+        )
+
+    db = get_database()
+    credentials = load_testing_username_password_credentials()
+    existing_credential = credentials.get(username)
+    user_id = ObjectId(existing_credential["user_id"]) if existing_credential else ObjectId()
+    existing_user = db.users.find_one({"_id": user_id})
+
+    user_update = {"account_status": status_value.value}
+    if display_name is not None or existing_user is None:
+        user_update["display_name"] = display_name or username
+    if contact_email is not None:
+        user_update["contact_email"] = contact_email
+
+    if existing_user is None:
+        user = Person(immutable_id=user_id, **user_update)
+        user_document = user.dict(by_alias=True, exclude_none=True)
+        user_document.pop("identities", None)
+        db.users.insert_one(user_document)
+    else:
+        db.users.update_one({"_id": user_id}, {"$set": user_update})
+
+    db.roles.update_one({"_id": user_id}, {"$set": {"role": role_value.value}}, upsert=True)
+    set_testing_username_password_credential(username, str(user_id), password)
+    print(f"Testing username/password user {username!r} is linked to {user_id}.")
+
+
+dev.add_task(create_testing_username_password_user)
+
+
 # Example data for `dev.seed`, grouped by the block that renders it. Each group
 # is expanded by `_build_seed_items` into one seeded item per file (a single
 # block per item), so a fresh dev instance exercises every block type that has