User/Group Seed File about lldap HOT 3 CLOSED

import subprocess

from requests import Session
from qlient.http import HTTPBackend, HTTPClient, Fields

USER=""
PASS=""
SERVER_URL=""

SEED = {
  "groups": ["group_1"],
  "users": [
    {
      "id": "service_user_a",
      "email": "[email protected]",
      "displayName": "Service A",
      "groups": ["lldap_password_manager"],
      "password_file": "./pass_a",
    },
    {
      "id": "server_user_b",
      "email": "[email protected]",
      "displayName": "Service B",
      "groups": ["lldap_strict_readonly"],
      "password_file": "./pass_b",
    },
  ],
}

session = Session()

jwt_token = session.post(url = f"{SERVER_URL}/auth/simple/login", json={
  "username": USER,
  "password": PASS,
}).json()["token"]

session.headers["Authorization"] = f"Bearer {jwt_token}"

client = HTTPClient(HTTPBackend(f"{SERVER_URL}/api/graphql", session=session))

GROUP_FIELDS = Fields("id", "displayName")
USER_FIELDS = Fields("id", "email", "displayName", groups=GROUP_FIELDS)

existing_groups = client.query.groups(GROUP_FIELDS).data["groups"]
existing_groups_map = {group["displayName"]: group for group in existing_groups}


def create_group(name: str):
  group = client.mutation.createGroup(name=name, _fields=GROUP_FIELDS).data["createGroup"]
  existing_groups[name] = group

  print(f"\tCreated group '{name}'.")

  return group
  

def create_user(user_original):
  user = user_original.copy()
  del user_original
  groups = user.pop("groups", [])
  password_file = user.pop("password_file")
  with open(password_file, "r") as f:
    password = f.read()

  # create user
  req = client.mutation.createUser(user=user, _fields=USER_FIELDS)
  assert req.errors is None, req.raw

  # add to groups
  for group in groups:
    req = client.mutation.addUserToGroup(userId=user["id"], groupId=group["id"])
    assert req.errors is None, req.raw

  # set password
  ret = subprocess.run([
    "lldap_set_password",
    "--base-url", SERVER_URL,
    "--username", user["id"],
    "--password", password,
    "--token", jwt_token,
    ], capture_output=True)
  ret.check_returncode()

  print("\tCreated user '{}'".format(user["id"]))


for must_exist_group in sorted(SEED.get("groups", [])):
  if must_exist_group not in existing_groups_map:
    print(f"Group '{must_exist_group}' does not exist, creating...")
    create_group(must_exist_group)
  else:
    print(f"Group '{must_exist_group}' exists, skipping")


failed = False
existing_users = client.query.users(USER_FIELDS).data["users"]
existing_users_map = { user["id"]: user for user in existing_users}

for must_exist_user in sorted(SEED.get("users", []), key=lambda x: x["id"]):
  must_exist_user_id = must_exist_user["id"]

  # Seed file groups is just a list of names, API returns id + displayName
  # TODO: This modifies the SEED inplace
  must_exist_user["groups"][:] = [
    existing_groups_map[group_name] for group_name in must_exist_user["groups"]
  ]

  if must_exist_user_id not in existing_users_map:
    print(f"User '{must_exist_user_id}' does not exist, creating...")
    create_user(must_exist_user)

  else:
    print(f"User '{must_exist_user_id}' exists, verifying details match... ", end="")
    
    existing_user = existing_users_map[must_exist_user_id]
    if must_exist_user != existing_user:
      failed = True
      print("failed")
      for attribute in must_exist_user.keys():
        requested = must_exist_user[attribute]
        existing = existing_user[attribute]
        if requested != existing:
          print(f"\tAttribute '{attribute}' does not match: (requested) '{requested}', (existing) '{existing}'.")
    else:
      print("ok")


if failed:
  print("\nUser details did not match... please fix these manually.")
  quit(1)