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/
 

src/fides/cli/__init__.py

@@ -19,6 +19,7 @@
 from .commands.scan import scan
 from .commands.util import deploy, init, status, webserver, worker
 from .commands.view import view
+from .commands.user import user
 
 CONTEXT_SETTINGS = dict(help_option_names=["-h", "--help"])
 LOCAL_COMMANDS = [deploy, evaluate, generate, init, scan, parse, view, webserver]
@@ -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,101 @@
+"""Contains the user group of commands for fides."""
+
+import click
+from typing import Dict, List
+
+from fides.core.user import (
+    create_user,
+    get_access_token,
+    write_credentials_file,
+    CREDENTIALS_PATH,
+    get_user_scopes,
+    create_auth_header,
+    read_credentials_file,
+)
+from fides.core.utils import echo_green
+
+
+@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 to get a user access token and write a credentials file."""
+
+    # If no username/password was provided, attempt to use the root
+    config = ctx.obj["CONFIG"]
+    client_id = config.security.oauth_root_client_id
+    client_secret = config.security.oauth_root_client_secret
+    access_token = get_access_token(client_id, client_secret)
+    auth_header = create_auth_header(access_token)
+
+    create_user(username, password, first_name, last_name, auth_header)
+
+
+@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 was provided, attempt to use the root
+    config = ctx.obj["CONFIG"]
+    username = username or config.security.oauth_root_client_id
+    password = password or config.security.oauth_root_client_secret
+
+    access_token = get_access_token(username, password)
+    write_credentials_file(username, access_token)
+    echo_green(f"Credentials file written to: {CREDENTIALS_PATH}")
+
+
+@user.command()
+@click.option(
+    "-u",
+    "--username",
+    default="",
+    help="Username to authenticate with.",
+)
+def scopes(username: str) -> None:
+    """List the scopes avaible to the current user."""
+
+    credentials: Dict[str, str] = read_credentials_file()
+    username = credentials["username"]
+    access_token = credentials["access_token"]
+
+    auth_header = create_auth_header(access_token)
+    scopes: List[str] = get_user_scopes(username, auth_header)
+    for scope in scopes:
+        print(scope)

src/fides/core/user.py

@@ -0,0 +1,93 @@
+"""Module for interaction with User endpoints/commands."""
+from pathlib import Path
+from typing import Dict, List
+
+import requests
+import toml
+
+from fides.core.utils import echo_red
+
+OAUTH_TOKEN_URL = "http://localhost:8080/api/v1/oauth/token"
+CLIENT_SCOPES_URL = "http://localhost:8080/api/v1/oauth/client/{}/scope"
+CREDENTIALS_PATH = f"{str(Path.home())}/.fides_credentials"
+
+
+def get_access_token(username: str, password: str) -> str:
+    """
+    Get a user access token from the webserver.
+    """
+    payload = {
+        "client_id": username,
+        "client_secret": password,
+        "grant_type": "client_credentials",
+    }
+
+    response = requests.post(OAUTH_TOKEN_URL, data=payload)
+    if response.status_code == 401:
+        echo_red(
+            "Authentication failed! Please check your username/password and try again."
+        )
+        raise SystemExit(1)
+    access_token = response.json()["access_token"]
+    return access_token
+
+
+def write_credentials_file(username: str, access_token: str) -> str:
+    """
+    Write the user credentials file.
+    """
+    credentials_data = {"username": username, "access_token": access_token}
+    with open(CREDENTIALS_PATH, "w", encoding="utf-8") as credentials_file:
+        credentials_file.write(toml.dumps(credentials_data))
+    return CREDENTIALS_PATH
+
+
+def read_credentials_file() -> Dict[str, str]:
+    """Read and return the credentials file."""
+    with open(CREDENTIALS_PATH, "r", encoding="utf-8") as credentials_file:
+        credentials_data = toml.load(credentials_file)
+    return credentials_data
+
+
+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 get_user_scopes(username: str, auth_header: Dict[str, str]) -> List[str]:
+    """
+    Get a user access token from the webserver.
+    """
+    scopes_url = CLIENT_SCOPES_URL.format(username)
+    response = requests.get(
+        scopes_url,
+        headers=auth_header,
+    )
+
+    if response.status_code != 200:
+        echo_red("Request failed! Please check your username/password and try again.")
+        raise SystemExit(1)
+
+    return response.json()
+
+
+def create_user(
+    username: str,
+    password: str,
+    first_name: str,
+    last_name: str,
+    auth_header: Dict[str, str],
+) -> None:
+    """Create a user."""
+    user_data = {
+        "username": username,
+        "password": password,
+        "first_name": first_name,
+        "last_name": last_name,
+    }
+    print(user_data)
+    response = requests.post(
+        "http://localhost:8080/api/v1/user", headers=auth_header, data=user_data
+    )
+    print(response.text)