123 lines
4.6 KiB
Python
123 lines
4.6 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
from collections.abc import Sequence
|
|
from datetime import timedelta
|
|
from getpass import getpass
|
|
|
|
from cloud.control_config import load_control_config
|
|
from cloud.database import CloudDatabase
|
|
from cloud.schema import require_current_schema
|
|
from cloud.user_auth import UserAuthService, UserAuthSettings, normalize_username, utc_now
|
|
|
|
|
|
def main(argv: Sequence[str] | None = None) -> None:
|
|
parser = argparse.ArgumentParser(description="Administer Cloud Console user accounts")
|
|
commands = parser.add_subparsers(dest="command", required=True)
|
|
users = commands.add_parser("users", help="manage user accounts")
|
|
user_commands = users.add_subparsers(dest="user_command", required=True)
|
|
|
|
create = user_commands.add_parser("create", help="create a user interactively")
|
|
create.add_argument("--username", required=True)
|
|
create.add_argument("--display-name")
|
|
create.add_argument("--role", choices=("viewer", "operator", "admin"), required=True)
|
|
|
|
reset = user_commands.add_parser("reset-password", help="reset a user password")
|
|
reset.add_argument("--username", required=True)
|
|
|
|
enable = user_commands.add_parser("enable", help="enable a disabled user")
|
|
enable.add_argument("--username", required=True)
|
|
|
|
revoke = user_commands.add_parser("revoke-sessions", help="revoke a user's sessions")
|
|
revoke.add_argument("--username", required=True)
|
|
|
|
args = parser.parse_args(argv)
|
|
config = load_control_config()
|
|
require_current_schema(config.database_url)
|
|
database = CloudDatabase(config.database_url, create_schema=False)
|
|
service = UserAuthService(
|
|
database.repository,
|
|
settings=UserAuthSettings(
|
|
session_idle_ttl=timedelta(seconds=config.user_session_idle_seconds),
|
|
session_absolute_ttl=timedelta(
|
|
seconds=config.user_session_absolute_seconds
|
|
),
|
|
login_failure_limit=config.login_failure_limit,
|
|
login_failure_window=timedelta(
|
|
seconds=config.login_failure_window_seconds
|
|
),
|
|
login_block_duration=timedelta(seconds=config.login_block_seconds),
|
|
cookie_secure=config.session_cookie_secure,
|
|
),
|
|
)
|
|
try:
|
|
_run_user_command(args, service)
|
|
finally:
|
|
database.close()
|
|
|
|
|
|
def _run_user_command(args: argparse.Namespace, service: UserAuthService) -> None:
|
|
username = normalize_username(args.username)
|
|
if args.user_command == "create":
|
|
password = _read_password()
|
|
user = service.create_user(
|
|
username=args.username,
|
|
display_name=args.display_name or args.username,
|
|
role=args.role,
|
|
password=password,
|
|
)
|
|
service.record_admin_action(
|
|
actor_principal_id="deployment-cli",
|
|
target_user_id=user.id,
|
|
action="user_create",
|
|
metadata={"role": user.role},
|
|
)
|
|
print(f"created user {user.username!r} with role {user.role}")
|
|
return
|
|
|
|
user = service.repository.get_user_by_normalized_username(username) # type: ignore[attr-defined]
|
|
if user is None:
|
|
raise SystemExit("user not found")
|
|
if args.user_command == "reset-password":
|
|
service.reset_password(
|
|
user_id=user.id,
|
|
new_password=_read_password(),
|
|
actor_principal_id="deployment-cli",
|
|
)
|
|
print(f"reset password for {user.username!r}")
|
|
return
|
|
if args.user_command == "enable":
|
|
updated = service.repository.update_user( # type: ignore[attr-defined]
|
|
user.id,
|
|
enabled=True,
|
|
updated_at=utc_now(),
|
|
)
|
|
service.record_admin_action(
|
|
actor_principal_id="deployment-cli",
|
|
target_user_id=updated.id,
|
|
action="user_enable",
|
|
)
|
|
print(f"enabled user {updated.username!r}")
|
|
return
|
|
if args.user_command == "revoke-sessions":
|
|
service.repository.revoke_user_sessions( # type: ignore[attr-defined]
|
|
user.id,
|
|
revoked_at=utc_now(),
|
|
)
|
|
service.record_admin_action(
|
|
actor_principal_id="deployment-cli",
|
|
target_user_id=user.id,
|
|
action="session_revoke",
|
|
)
|
|
print(f"revoked sessions for {user.username!r}")
|
|
return
|
|
raise AssertionError(f"unsupported command {args.user_command!r}")
|
|
|
|
|
|
def _read_password() -> str:
|
|
password = getpass("Password: ")
|
|
confirmation = getpass("Confirm password: ")
|
|
if password != confirmation:
|
|
raise SystemExit("password confirmation did not match")
|
|
return password
|