Add CLI user commands

.gitignore

@@ -1,7 +1,5 @@
 # Source for the following rules: https://raw.githubusercontent.com/github/gitignore/master/Python.gitignore
 
-# Files to keep that would otherwise get ignored
-
 # frontend
 ui-build/
 

pyproject.toml

@@ -155,10 +155,8 @@ log_cli=false
 filterwarnings = "ignore::DeprecationWarning:aiofiles.*:"
 testpaths="tests"
 log_level = "INFO"
-addopts = ["--cov=fides.core",
-            "--cov=fides.cli",
-            "--cov=fides.api",
-            "--cov-report=term-missing",
+addopts = ["--cov=fides",
+            "-ra",
             "-vv",
             "--no-cov-on-fail",
             "--disable-pytest-warnings"]

src/fides/cli/__init__.py

@@ -17,6 +17,7 @@
 from .commands.export import export
 from .commands.generate import generate
 from .commands.scan import scan
+from .commands.user import user
 from .commands.util import deploy, init, status, webserver, worker
 from .commands.view import view
 
@@ -36,6 +37,7 @@
     pull,
     push,
     worker,
+    user,
 ]
 API_COMMAND_DICT = {command.name or str(command): command for command in API_COMMANDS}
 ALL_COMMANDS = API_COMMANDS + LOCAL_COMMANDS

src/fides/cli/commands/user.py

@@ -0,0 +1,135 @@
+"""Contains the user group of commands for fides."""
+
+from typing import List
+
+import click
+
+from fides.core.user import (
+    create_auth_header,
+    create_user,
+    get_access_token,
+    get_user_permissions,
+    read_credentials_file,
+    update_user_permissions,
+    write_credentials_file,
+)
+from fides.core.utils import echo_green, echo_red
+from fides.lib.oauth.scopes import SCOPES
+
+
+@click.group(name="user")
+@click.pass_context
+def user(ctx: click.Context) -> None:
+    """
+    Click command group for interacting with user-related functionality.
+    """
+
+
+@user.command()
+@click.pass_context
+@click.argument("username", type=str)
+@click.argument("password", type=str)
+@click.option(
+    "-f",
+    "--first-name",
+    default="",
+    help="First name of the user.",
+)
+@click.option(
+    "-l",
+    "--last-name",
+    default="",
+    help="Last name of the user.",
+)
+def create(
+    ctx: click.Context, username: str, password: str, first_name: str, last_name: str
+) -> None:
+    """
+    Use credentials from the credentials file to create a new user.
+
+    Gives full permissions.
+    """
+    config = ctx.obj["CONFIG"]
+    server_url = config.cli.server_url
+
+    try:
+        credentials = read_credentials_file()
+    except FileNotFoundError:
+        echo_red("No credentials file found.")
+        raise SystemExit(1)
+
+    access_token = credentials.access_token
+    auth_header = create_auth_header(access_token)
+    user_response = create_user(
+        username=username,
+        password=password,
+        first_name=first_name,
+        last_name=last_name,
+        auth_header=auth_header,
+        server_url=server_url,
+    )
+    user_id = user_response.json()["id"]
+    update_user_permissions(
+        user_id=user_id, scopes=SCOPES, auth_header=auth_header, server_url=server_url
+    )
+    echo_green(f"User: '{username}' created and assigned permissions.")
+
+
+@user.command()
+@click.pass_context
+@click.option(
+    "-u",
+    "--username",
+    default="",
+    help="Username to authenticate with.",
+)
+@click.option(
+    "-p",
+    "--password",
+    default="",
+    help="Password to authenticate with.",
+)
+def login(ctx: click.Context, username: str, password: str) -> None:
+    """
+    Use credentials to get a user access token and write a credentials file.
+
+    If no username/password is provided, attempt to load an existing credentials file
+    and use that username/password.
+    """
+    config = ctx.obj["CONFIG"]
+    server_url = config.cli.server_url
+
+    if not (username and password):
+        try:
+            credentials = read_credentials_file()
+        except FileNotFoundError:
+            echo_red("No username/password provided and no credentials file found.")
+            raise SystemExit(1)
+        username = credentials.username
+        password = credentials.password
+
+    user_id, access_token = get_access_token(
+        username=username, password=password, server_url=server_url
+    )
+    echo_green(f"Logged in as user: {username}")
+    credentials_path = write_credentials_file(username, password, user_id, access_token)
+    echo_green(f"Credentials file written to: {credentials_path}")
+
+
+@user.command(name="permissions")
+@click.pass_context
+def get_permissions(ctx: click.Context) -> None:
+    """List the scopes avaible to the current user."""
+    config = ctx.obj["CONFIG"]
+    server_url = config.cli.server_url
+
+    credentials = read_credentials_file()
+    user_id = credentials.user_id
+    access_token = credentials.access_token
+
+    auth_header = create_auth_header(access_token)
+    permissions: List[str] = get_user_permissions(user_id, auth_header, server_url)
+
+    print("Permissions:")
+    for permission in permissions:
+        print(f"\t{permission}")

src/fides/core/user.py

@@ -0,0 +1,135 @@
+"""Module for interaction with User endpoints/commands."""
+import json
+from pathlib import Path
+from typing import Dict, List, Tuple
+
+import requests
+import toml
+from pydantic import BaseModel
+
+from fides.cli.utils import handle_cli_response
+from fides.core.config import get_config
+from fides.lib.cryptography.cryptographic_util import str_to_b64_str
+
+config = get_config()
+CREATE_USER_PATH = "/api/v1/user"
+LOGIN_PATH = "/api/v1/login"
+USER_PERMISSIONS_PATH = "/api/v1/user/{}/permission"
+
+CREDENTIALS_FILE_PATH = f"{str(Path.home())}/.fides_credentials"
+
+
+class Credentials(BaseModel):
+    """
+    User credentials for the CLI.
+    """
+
+    username: str
+    password: str
+    user_id: str
+    access_token: str
+
+
+def get_access_token(username: str, password: str, server_url: str) -> Tuple[str, str]:
+    """
+    Get a user access token from the webserver.
+    """
+    payload = {
+        "username": username,
+        "password": str_to_b64_str(password),
+    }
+
+    response = requests.post(server_url + LOGIN_PATH, json=payload)
+    handle_cli_response(response, verbose=False)
+    user_id: str = response.json()["user_data"]["id"]
+    access_token: str = response.json()["token_data"]["access_token"]
+    return (user_id, access_token)
+
+
+def write_credentials_file(
+    username: str, password: str, user_id: str, access_token: str
+) -> str:
+    """
+    Write the user credentials file.
+    """
+    credentials = Credentials(
+        username=username,
+        password=password,
+        user_id=user_id,
+        access_token=access_token,
+    )
+    with open(CREDENTIALS_FILE_PATH, "w", encoding="utf-8") as credentials_file:
+        credentials_file.write(toml.dumps(credentials.dict()))
+    return CREDENTIALS_FILE_PATH
+
+
+def read_credentials_file() -> Credentials:
+    """Read and return the credentials file."""
+    with open(CREDENTIALS_FILE_PATH, "r", encoding="utf-8") as credentials_file:
+        credentials = Credentials.parse_obj(toml.load(credentials_file))
+    return credentials
+
+
+def create_auth_header(access_token: str) -> Dict[str, str]:
+    """Given an access token, create an auth header."""
+    auth_header = {
+        "Authorization": f"Bearer {access_token}",
+    }
+    return auth_header
+
+
+def create_user(
+    username: str,
+    password: str,
+    first_name: str,
+    last_name: str,
+    auth_header: Dict[str, str],
+    server_url: str,
+) -> requests.Response:
+    """Create a user."""
+    request_data = {
+        "username": username,
+        "password": str_to_b64_str(password),
+        "first_name": first_name,
+        "last_name": last_name,
+    }
+    response = requests.post(
+        server_url + CREATE_USER_PATH,
+        headers=auth_header,
+        data=json.dumps(request_data),
+    )
+    handle_cli_response(response, verbose=False)
+    return response
+
+
+def get_user_permissions(
+    user_id: str, auth_header: Dict[str, str], server_url: str
+) -> List[str]:
+    """
+    List all of the user permissions for the provided user.
+    """
+    get_permissions_path = USER_PERMISSIONS_PATH.format(user_id)
+    response = requests.get(
+        server_url + get_permissions_path,
+        headers=auth_header,
+    )
+
+    handle_cli_response(response, verbose=False)
+    return response.json()["scopes"]
+
+
+def update_user_permissions(
+    user_id: str, scopes: List[str], auth_header: Dict[str, str], server_url: str
+) -> requests.Response:
+    """
+    Update user permissions for a given user.
+    """
+    request_data = {"scopes": scopes, "id": user_id}
+    set_permissions_path = USER_PERMISSIONS_PATH.format(user_id)
+    response = requests.put(
+        server_url + set_permissions_path,
+        headers=auth_header,
+        json=request_data,
+    )
+    handle_cli_response(response, verbose=False)
+    return response