elab-users/elab_users/__init__.py

""" Elab Users

Manage elab (svn) users
"""

__version__ = "0.0.1"

import os
import sys
import argparse
import subprocess  # noqa: S404
from pathlib import Path

from .authz import AuthzConfigParser
from .constants import (
    USERS,
    ADMINS,
    ALUMNI,
    READ_ACL,
    WRITE_ACL,
    RESTRICTED,
    SVN_SUFFIX,
    AUTHZ_FILE_NAME,
    HTPWD_FILE_NAME,
)

SVN_REPOS_PATH = Path(os.getenv("SVN_REPOS_PATH", default=".")).resolve()

COMMANDS = ["user", "groups", "add", "restricted", "retire", "password"]


def get_config(svn_dir=SVN_REPOS_PATH, authz=AUTHZ_FILE_NAME):
    authz_path = Path(svn_dir) / authz
    if not authz_path.is_file():
        sys.exit(f"Could not find authz file at {authz_path}")
    return AuthzConfigParser.from_file(authz_path)


def list_users(svn_dir=SVN_REPOS_PATH, authz=AUTHZ_FILE_NAME):
    """list all users"""
    config = get_config(svn_dir, authz)
    groups = config.group_users()
    for name, usernames in groups.items():
        print(f"Users in group '{name}':")
        for name in sorted(usernames):
            print(f"  {name}")


def show_group_info(groupname, svn_dir=SVN_REPOS_PATH, authz=AUTHZ_FILE_NAME):
    """show group information"""
    config = get_config(svn_dir, authz)
    groups = config.group_users()
    if groupname not in groups:
        sys.exit(f"Group '{groupname}' not found in authz file")
    print(f"Users in group '{groupname}':")
    for name in sorted(groups[groupname]):
        print(f"  {name}")


def add_new_user(
    username,
    group,
    svn_dir=SVN_REPOS_PATH,
    authz=AUTHZ_FILE_NAME,
    htpwd=HTPWD_FILE_NAME,
    handler=subprocess,
):
    """add a user, restricted or regular"""
    config = get_config(svn_dir, authz)
    if username in config.elab_users:
        sys.exit(f"Username '{username}' already in use")
    if username.lower() in {i.name.lower() for i in svn_dir.iterdir()}:
        sys.exit(f"Username '{username}' not allowed")
    user = config.add_journal_acl_for(username, group)
    user.create_new_repository(svn_dir, handler)
    password = user.set_new_password(svn_dir / htpwd, handler=handler)
    print("New password for :")
    print(f"username: {username}")
    print(f"password: {password}")
    print(f"url:      https://svn.cpi.imtek.uni-freiburg.de/{username}")
    config.write_to_file()


def retire_user(
    username,
    svn_dir=SVN_REPOS_PATH,
    authz=AUTHZ_FILE_NAME,
    htpwd=HTPWD_FILE_NAME,
    handler=subprocess,
):
    config = get_config(svn_dir, authz)
    if username not in config.elab_users:
        sys.exit("User {username} not found.")
    user = config.elab_users[username]
    if user.group == ALUMNI:
        sys.exit(f"User '{username}' is already in group '{ALUMNI}'")
    if user.group == ADMINS:
        sys.exit(
            (
                f"User '{username}' is in group '{ADMINS}', "
                f"will not moved to '{ALUMNI}'"
            )
        )
    config.move_user_to_alumni(username)
    config.write_to_file()
    user.delete_password(svn_dir / htpwd, handler=handler)
    print(f"Moved user {username} to alumni")


def change_password(
    username,
    svn_dir=SVN_REPOS_PATH,
    authz=AUTHZ_FILE_NAME,
    htpwd=HTPWD_FILE_NAME,
    handler=subprocess,
):
    config = get_config(svn_dir, authz)
    if username not in config.elab_users:
        sys.exit("User {username} not found.")
    user = config.elab_users[username]
    password = user.set_new_password(svn_dir / htpwd, handler=handler)
    print("New password for :")
    print(f"username: {username}")
    print(f"password:  {password}")


def show_user_info(username, svn_dir=SVN_REPOS_PATH, authz=AUTHZ_FILE_NAME):
    config = get_config(svn_dir, authz)
    if username not in config.elab_users:
        sys.exit("User {username} not found.")

    user = config.elab_users[username]
    print(f"User {user.name} is in group '{user.group}':")

    # print the write acls for a user
    if user.group == ADMINS:
        print("  Write access is granted to all journals.")
    elif user.write_acl:
        write_acl = [item + SVN_SUFFIX for item in user.write_acl]
        print("  Write access is granted to:", ", ".join(write_acl))
    else:
        print("  Write access is NOT granted to any journals")

    # print the read acls for a user
    if user.group == ADMINS:
        print("  Read access is granted to all journals.")
    elif user.group == USERS:
        print("  Read access is granted to (nearly) all journals.")
    elif user.read_acl:
        read_acl = [item + SVN_SUFFIX for item in user.read_acl]
        print("  Read access is granted to:", ", ".join(read_acl))
    else:
        print("  Read access is NOT granted to any journals")

    journal = config.get_journal_info(username)
    print(f"Labjournal {username}{SVN_SUFFIX}")

    # print the write acls for a journal
    if journal[WRITE_ACL]:
        print("  Write access granted to:", ", ".join(journal[WRITE_ACL]))
    else:
        print("  No write access granted to anybody")

    # print the read acls for a journal
    if journal[READ_ACL]:
        print("  Read access granted to:", ", ".join(journal[READ_ACL]))
    else:
        print("  No read access granted to anybody")


def main(
    svn_dir=SVN_REPOS_PATH,
    authz=AUTHZ_FILE_NAME,
    htpwd=HTPWD_FILE_NAME,
    handler=subprocess,
    cli_args=None,
):
    parser = argparse.ArgumentParser(prog="elab-users")
    parser.add_argument(
        "command",
        nargs="?",
        help="one of the commands: [" + ", ".join(COMMANDS) + "]",
    )
    parser.add_argument(
        "name", nargs="?", help="user or group to perform the command on"
    )
    args = parser.parse_args(cli_args)
    print(args.command)
    print(args.name)

    if not args.command:
        list_users(svn_dir, authz)
    elif args.command.lower() not in COMMANDS:
        show_user_info(args.command, svn_dir, authz)
    elif args.command.lower() == "user":
        show_user_info(args.name, svn_dir, authz)
    elif args.command.lower() == "groups":
        show_group_info(args.name, svn_dir, authz)
    elif args.command.lower() == "add":
        add_new_user(args.name, USERS, svn_dir, authz, htpwd, handler)
    elif args.command.lower() == "restricted":
        add_new_user(args.name, RESTRICTED, svn_dir, authz, htpwd, handler)
    elif args.command.lower() == "retire":
        retire_user(args.name, svn_dir, authz, htpwd, handler)
    elif args.command.lower() == "password":
        change_password(args.name, svn_dir, authz, htpwd, handler)