Add ability to create test users with magic link logins

pydatalab/docs/INSTALL.md

@@ -215,7 +215,7 @@ uv lock
 ```
 ### Test server authentication/authorisation
 
-There are two approaches to authentication when developing *datalab* features locally.
+There are three approaches to authentication when developing *datalab* features locally.
 
 1. Disable authentication entirely with the `PYDATALAB_TESTING=true` environment
    variable (or corresponding config file option `TESTING`). This will perform
@@ -224,7 +224,7 @@ There are two approaches to authentication when developing *datalab* features lo
    - This mode of development is fine for e.g., developing new blocks, but in
      cases where new API functionality is being added, it is recommended to set
      up authentication locally (see below).
-1. Local OAuth setup. This requires registering an OAuth app with one of the
+2. Local OAuth setup. This requires registering an OAuth app with one of the
    implemented providers (e.g., GitHub, ORCID), configuring the credentials
    locally (see the [configuration documentation](https://docs.datalab-org.io/en/latest/config/) for more details) and then logging into *datalab* normally.
    - In this case, the user will also need to be activated when it is created.
@@ -233,6 +233,30 @@ There are two approaches to authentication when developing *datalab* features lo
      invoke task.
    - For testing admin functionality, the user can also be promoted with
      the `admin.change-user-role` invoke task.
+3. Test users with magic-link login URLs, which requires the server to run in testing mode
+   (`uv run invoke dev.serve --testing`, or `PYDATALAB_TESTING=true`). Create some test users (active accounts
+   with random `@datalab.test` email addresses) with:
+
+   ```shell
+   uv run invoke dev.create-test-user --count 3
+   uv run invoke dev.create-test-user --role admin
+   ```
+
+   A specific user can be created or updated with
+   `--username alice --display-name "Alice" --role manager`.
+
+   Then list the test users with their roles, groups and login links:
+
+   ```shell
+   uv run invoke dev.list-test-users
+   ```
+
+   Each link is a login token (valid for one hour), minted directly instead of
+   being sent by email, pointing at `PYDATALAB_APP_URL`. Open each link in a
+   separate private browser window to test roles, groups and permissions as
+   several users at once. The server only accepts these links for `@datalab.test`
+   users; as this domain cannot receive email, they cannot be used to log in as a
+   real user.
 
 Finally, all API tests can be run with variable authentication.
 There are [pytest fixtures](https://docs.pytest.org/en/7.1.x/how-to/fixtures.html) that provide

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

@@ -33,6 +33,12 @@
 
 KEY_LENGTH: int = 32
 LINK_EXPIRATION: datetime.timedelta = datetime.timedelta(hours=1)
+
+TESTING_EMAIL_DOMAIN: str = "datalab.test"
+"""The reserved domain (RFC 2606) for test users, which are the only users that can log in
+with magic-link tokens minted outside of email (e.g., by invoke tasks), and only when
+`CONFIG.TESTING` is enabled. As the domain cannot receive email, these accounts can never
+belong to a real person."""
 OAUTH_NEXT_SESSION_KEY_PREFIX = "oauth_next_"
 OAUTH_NEXT_MAX_LENGTH = 2048
 REMEMBER_ME_SESSION_KEY = "remember_me"
@@ -676,7 +682,12 @@ def _validate_magic_link_request(email: str, referrer: str) -> None:
         raise BadRequest("Referrer address not provided, please contact the datalab administrator")
 
 
-def _generate_and_store_token(email: str, intent: str = "register", remember: bool = False) -> str:
+def _generate_and_store_token(
+    email: str,
+    intent: str = "register",
+    remember: bool = False,
+    channel: str = "email",
+) -> str:
     """Generate a JWT for the user with a short expiration and store it in the session.
 
     The session itself persists beyond the JWT expiration. The `exp` key is a standard
@@ -687,6 +698,9 @@ def _generate_and_store_token(email: str, intent: str = "register", remember: bo
         intent: The intent of the magic link, e.g., "register" "verify", or "login".
         remember: Whether the session created by the magic link should persist
             beyond the browser session.
+        channel: How the token is delivered to the user: "email", or "cli" for tokens
+            minted directly by invoke tasks, which can only be redeemed by test users
+            when `CONFIG.TESTING` is enabled.
 
     Returns:
         The generated JWT token string.
@@ -697,6 +711,7 @@ def _generate_and_store_token(email: str, intent: str = "register", remember: bo
         "email": email,
         "intent": intent,
         "remember": remember,
+        "channel": channel,
     }
 
     token = jwt.encode(
@@ -705,7 +720,14 @@ def _generate_and_store_token(email: str, intent: str = "register", remember: bo
         algorithm="HS256",
     )
 
-    flask_mongo.db.magic_links.insert_one({"jwt": token})
+    flask_mongo.db.magic_links.insert_one(
+        {
+            "jwt": token,
+            "channel": channel,
+            "email": email,
+            "created_at": datetime.datetime.now(datetime.timezone.utc),
+        }
+    )
 
     return token
 
@@ -874,6 +896,17 @@ def email_logged_in():
     if not email:
         raise BadRequest("No email found; please request a new token.")
 
+    # Tokens issued before the `channel` claim was added were all sent by email
+    channel = data.get("channel", "email")
+    if channel != "email":
+        if not CONFIG.TESTING or not email.endswith(f"@{TESTING_EMAIL_DOMAIN}"):
+            LOGGER.warning("Rejected %r login token for %s", channel, email)
+            raise Forbidden(
+                f"Login tokens not sent by email are only accepted for @{TESTING_EMAIL_DOMAIN} "
+                "users when the server is in testing mode."
+            )
+    LOGGER.info("Magic-link login for %s via %r token", email, channel)
+
     # A magic link always logs in as the owner of the email; log out any current user
     # so that the email is never attached to their account as a new identity
     if current_user.is_authenticated:

pydatalab/tasks.py

@@ -4,6 +4,7 @@
 import re
 import shutil
 import subprocess
+import sys
 import time
 
 import tomlkit
@@ -137,6 +138,30 @@ def serve(
     debug: bool = False,
 ):
     """Boot the Flask development server."""
+    env_path = _load_dev_env()
+
+    if testing and "PYDATALAB_TESTING" not in os.environ:
+        os.environ["PYDATALAB_TESTING"] = "1"
+
+    if debug and "PYDATALAB_DEBUG" not in os.environ:
+        os.environ["PYDATALAB_DEBUG"] = "1"
+
+    from pydatalab.main import create_app
+
+    create_app(env_file=env_path).run(host=host, port=port, debug=debug, use_reloader=reload)
+
+
+dev.add_task(serve)
+
+
+def _load_dev_env() -> pathlib.Path:
+    """Load the development `.env` file and apply the insecure dev secret key fallback.
+
+    This must be called before importing anything that instantiates `CONFIG`, and is
+    shared between `dev.serve` and the test-user tasks so that login tokens minted in
+    the terminal are signed with the same key as the running dev server.
+
+    """
     from dotenv import load_dotenv
 
     # Load .env into os.environ *first*, before the guards below and before
@@ -151,22 +176,215 @@ def serve(
 
     load_dotenv(env_path)
 
-    if testing and "PYDATALAB_TESTING" not in os.environ:
-        os.environ["PYDATALAB_TESTING"] = "1"
-
-    if debug and "PYDATALAB_DEBUG" not in os.environ:
-        os.environ["PYDATALAB_DEBUG"] = "1"
-
     if "PYDATALAB_SECRET_KEY" not in os.environ:
         os.environ["PYDATALAB_SECRET_KEY"] = "dev-insecure-secret-key-do-not-use-in-production"  # noqa: S105
         os.environ["PYDATALAB_ALLOW_INSECURE_SECRET_KEY"] = "1"  # noqa: S105
 
+    return env_path
+
+
+@task(
+    help={
+        "username": "Username for the test user; their email will be <username>@datalab.test "
+        "(random if omitted)",
+        "display_name": "Display name for the test user (defaults to the username)",
+        "role": "User role: user, manager, or admin",
+        "count": "Number of users to create with random usernames (only without --username)",
+    }
+)
+def create_test_user(
+    _,
+    username: str | None = None,
+    display_name: str | None = None,
+    role: str = "user",
+    count: int = 1,
+):
+    """Create or update active test users that can log in via `dev.list-test-users` links."""
+
+    _load_dev_env()
+
+    import secrets
+
+    from pydantic import TypeAdapter, ValidationError
+
+    from pydatalab.models.people import AccountStatus, DisplayName, Identity, IdentityType, Person
+    from pydatalab.models.utils import HumanReadableIdentifier, UserRole
+    from pydatalab.mongo import get_database, insert_pydantic_model_fork_safe
+    from pydatalab.routes.v0_1.auth import TESTING_EMAIL_DOMAIN
+
+    try:
+        role_value = UserRole(role.lower())
+    except ValueError:
+        allowed_roles = ", ".join(value.value for value in UserRole)
+        raise SystemExit(f"Invalid role {role!r}; expected one of: {allowed_roles}.") from None
+
+    if display_name is not None:
+        try:
+            display_name = TypeAdapter(DisplayName).validate_python(display_name)
+        except ValidationError as exc:
+            raise SystemExit(f"Invalid display name {display_name!r}: {exc}") from None
+
+    if count < 1:
+        raise SystemExit("--count must be at least 1.")
+
+    database = get_database()
+
+    def find_user(email: str):
+        return database.users.find_one(
+            {"identities.identifier": email, "identities.identity_type": IdentityType.EMAIL.value}
+        )
+
+    if username is not None:
+        if count != 1:
+            raise SystemExit("--count cannot be combined with --username.")
+        try:
+            username = TypeAdapter(HumanReadableIdentifier).validate_python(username)
+        except ValidationError as exc:
+            raise SystemExit(f"Invalid test username {username!r}: {exc}") from None
+        new_users = [(username, display_name)]
+    else:
+        if count != 1 and display_name is not None:
+            raise SystemExit("--display-name cannot be combined with --count.")
+        new_users = [(secrets.token_hex(3), display_name) for _ in range(count)]
+
+    for username, user_display_name in new_users:
+        email = f"{username}@{TESTING_EMAIL_DOMAIN}"
+        existing_user = find_user(email)
+
+        if existing_user is None:
+            identity = Identity(
+                identity_type=IdentityType.EMAIL,
+                identifier=email,
+                name=email,
+                display_name=user_display_name or username,
+                verified=True,
+            )
+            user = Person.new_user_from_identity(
+                identity,
+                use_contact_email=False,
+                account_status=AccountStatus.ACTIVE,
+            )
+            user_id = insert_pydantic_model_fork_safe(user, "users")
+            action = "Created"
+        else:
+            user_id = existing_user["_id"]
+            user_updates = {"account_status": AccountStatus.ACTIVE.value}
+            if user_display_name is not None:
+                user_updates["display_name"] = user_display_name
+            database.users.update_one({"_id": user_id}, {"$set": user_updates})
+            action = "Updated"
+
+        database.roles.update_one(
+            {"_id": user_id},
+            {"$set": {"role": role_value.value}},
+            upsert=True,
+        )
+        print(f"{action} test user {email!r} with role {role_value.value!r}.")
+
+
+dev.add_task(create_test_user)
+
+
+@task
+def list_test_users(_):
+    """List test users with magic-link login URLs for the local webapp.
+
+    Test users are those with an @datalab.test email, as created by `dev.create-test-user`.
+    Each link is a login token (valid for one hour), minted directly rather than sent by
+    email, and only accepted by the server in testing mode. Open each in a separate
+    private browser window to be logged in as several users at once.
+
+    """
+
+    env_path = _load_dev_env()
+
+    from pydatalab.config import CONFIG
     from pydatalab.main import create_app
+    from pydatalab.models.people import IdentityType
+    from pydatalab.models.utils import UserRole
+    from pydatalab.mongo import get_database
+    from pydatalab.routes.v0_1.auth import TESTING_EMAIL_DOMAIN, _generate_and_store_token
+
+    if CONFIG.DISABLE_MAGIC_LINK_AUTH:
+        raise SystemExit("Test-user login links require magic-link auth to be enabled.")
+    if not CONFIG.APP_URL:
+        raise SystemExit("Test-user login links require PYDATALAB_APP_URL to point to the webapp.")
+
+    database = get_database()
+
+    email_suffix = f"@{TESTING_EMAIL_DOMAIN}"
+    documents = database.users.find(
+        {
+            "identities": {
+                "$elemMatch": {
+                    "identity_type": IdentityType.EMAIL.value,
+                    "identifier": {"$regex": f"{re.escape(email_suffix)}$"},
+                }
+            }
+        }
+    )
 
-    create_app(env_file=env_path).run(host=host, port=port, debug=debug, use_reloader=reload)
+    users = []
+    for document in documents:
+        email = next(
+            identity["identifier"]
+            for identity in document["identities"]
+            if identity.get("identity_type") == IdentityType.EMAIL.value
+            and identity.get("identifier", "").endswith(email_suffix)
+        )
+        role_document = database.roles.find_one({"_id": document["_id"]}, {"role": 1})
+        role = role_document["role"] if role_document else UserRole.USER.value
+        group_ids = [
+            group["immutable_id"]
+            for group in document.get("groups", [])
+            if group.get("immutable_id") is not None
+        ]
+        groups = database.groups.find(
+            {"_id": {"$in": group_ids}},
+            {"display_name": 1, "group_id": 1},
+        )
+        group_names = sorted(
+            (
+                group.get("display_name") or group.get("group_id") or str(group["_id"])
+                for group in groups
+            ),
+            key=str.casefold,
+        )
+        users.append(
+            {
+                "email": email,
+                "display_name": document.get("display_name") or email,
+                "role": role,
+                "status": document.get("account_status"),
+                "groups": group_names,
+            }
+        )
 
+    users.sort(key=lambda user: (user["display_name"].casefold(), user["email"]))
+    if not users:
+        print("No test users found; create one with `invoke dev.create-test-user`.")
+        return
 
-dev.add_task(serve)
+    use_color = sys.stdout.isatty() and "NO_COLOR" not in os.environ
+
+    def styled(value: str, code: str) -> str:
+        return f"\033[{code}m{value}\033[0m" if use_color and code else value
+
+    login_base_url = f"{CONFIG.APP_URL.rstrip('/')}/?token="
+    role_colors = {"admin": "31", "manager": "33", "user": "32"}
+    print(styled("Test users (links valid for 1 hour, with the server in testing mode)", "1;36"))
+    with create_app(env_file=env_path).app_context():
+        for user in users:
+            token = _generate_and_store_token(user["email"], intent="login", channel="cli")
+            groups = ", ".join(user["groups"]) if user["groups"] else "No groups"
+            print(f"\n{styled(user['display_name'], '1')} ({user['email']})")
+            print(f"  Role:   {styled(user['role'], role_colors.get(user['role'], ''))}")
+            print(f"  Status: {user['status']}")
+            print(f"  Groups: {groups}")
+            print(f"  Login:  {styled(login_base_url + token, '36')}")
+
+
+dev.add_task(list_test_users)
 
 
 @task