feat: add admin cli

labctl/commands/__init__.py

@@ -1,2 +1,3 @@
 from .config import app as config_app
 from .devices import app as devices_app
+from .admin import app as admin_app

labctl/commands/admin/__init__.py

@@ -0,0 +1,8 @@
+import typer
+from .users import app as users_app
+from .vpn import app as vpn_app
+
+app = typer.Typer()
+
+app.add_typer(users_app, name="users")
+app.add_typer(vpn_app, name="vpn")

labctl/commands/admin/users.py

@@ -0,0 +1,77 @@
+import typer
+
+from rich.table import Table
+
+from labctl.core import Config, APIDriver, console
+
+app = typer.Typer()
+
+@app.command(name="list")
+def list():
+    config = Config()
+    users = APIDriver().get("/users/").json()
+    table = Table(title=":adult: Users")
+    print(users)
+    table.add_column("Username", style="bold blue")
+    table.add_column("Email", style="cyan")
+    table.add_column("Disabled", style="yellow")
+    table.add_column("Is Admin", style="red")
+
+    for user in users:
+        table.add_row(
+            user["username"],
+            user["email"],
+            "Yes" if user["disabled"] else "No",
+            "Yes" if user["is_admin"] else "No",
+        )
+    console.print(table)
+
+@app.command(name="create")
+def create(username: str, email: str):
+    """
+    Create a new user
+    """
+    api_driver = APIDriver()
+    rsp = api_driver.post("/users/", json={"username": username, "email": email})
+    if rsp.status_code >= 400:
+        console.print(f"[red]Error: {rsp.text}[/red]")
+        return
+    console.print(f"User {username} created successfully")
+
+@app.command(name="show")
+def show(username: str):
+    """
+    Show the user details
+    """
+    user = APIDriver().get(f"/users/{username}").json()
+    vpn_groups = APIDriver().get(f"/users/{username}/vpn-group").json()
+    table = Table(title=":adult: User")
+    table.add_column("Key", style="cyan")
+    table.add_column("Value", style="magenta")
+    table.add_row("Username", user["username"])
+    table.add_row("Email", user["email"])
+    table.add_row("Disabled", "Yes" if user["disabled"] else "No")
+    table.add_row("Is Admin", "Yes" if user["is_admin"] else "No")
+    table.add_row("VPN Groups", ", ".join(vpn_groups.get("groups", [])))
+    console.print(table)
+
+@app.command(name="delete")
+def delete(username: str, confirm: bool = typer.Option(False, "--confirm", help="Confirm the deletion")):
+    """
+    Delete a user
+    """
+    if not confirm:
+        console.print("[red]Please confirm the deletion with --confirm[/red]")
+        return
+    rsp = APIDriver().delete(f"/users/{username}")
+    if rsp.status_code >= 400:
+        console.print(f"[red]Error: {rsp.text}[/red]")
+        return
+    console.print(f"User {username} deleted successfully")
+
+@app.command(name="sync")
+def sync():
+    """
+    Sync users from the directory
+    """
+    console.print("Todo")

labctl/commands/admin/vpn.py

@@ -0,0 +1,35 @@
+import typer
+
+from rich.table import Table
+
+from labctl.core import Config, APIDriver, console
+
+app = typer.Typer()
+
+app_group = typer.Typer()
+
+app.add_typer(app_group, name="group")
+
+@app_group.command(name="add-user")
+def add_user(username: str, group: str):
+    """
+    Add user to group
+    """
+    api_driver = APIDriver()
+    rsp = api_driver.post(f"/users/{username}/vpn-group/{group}")
+    if rsp.status_code >= 400:
+        console.print(f"[red]Error: {rsp.text}[/red]")
+        return
+    console.print(f"User {username} added to group {group}")
+
+@app_group.command(name="del-user")
+def del_user(username: str, group: str):
+    """
+    Delete user from group
+    """
+    api_driver = APIDriver()
+    rsp = api_driver.delete(f"/users/{username}/vpn-group/{group}")
+    if rsp.status_code >= 400:
+        console.print(f"[red]Error: {rsp.text}[/red]")
+        return
+    console.print(f"User {username} deleted from group {group}")

labctl/commands/config.py

@@ -29,6 +29,8 @@ def show():
     table.add_row("API URL", config.api_endpoint)
     table.add_row("API User", config.username)
     table.add_row("API Token", api_token)
+    if config.admin_cli:
+        table.add_row("Admin CLI Mode", "Enabled")
     console.print(table)
 
     if not config.api_endpoint:

labctl/core/config.py

@@ -1,16 +1,18 @@
 from pathlib import Path
+from os import environ
 
 import yaml
 
 CONFIG_LOCATION_DIR = f"{Path.home()}/"
-CONIIG_FILE = ".labctl_config.yaml"
+CONFIG_FILE = environ.get("LABCTL_CONFIG_FILE", ".labctl.yaml")
 
 class Config:
 
     api_endpoint: str
     username: str
     api_token: str
     token_type: str
+    admin_cli: bool = False
 
     def __init__(self, **kwargs):
         """
@@ -22,7 +24,7 @@ def __init__(self, **kwargs):
             Path(CONFIG_LOCATION_DIR).mkdir(parents=True)
 
         # If the config file does not exist, create a new one else load the existing one
-        if not Path(CONFIG_LOCATION_DIR + CONIIG_FILE).exists():
+        if not Path(CONFIG_LOCATION_DIR + CONFIG_FILE).exists():
             self.save()
         else:
             self.load()
@@ -38,20 +40,20 @@ def save(self):
         """
         Save the current configuration to the configuration file
         """
-        with open(CONFIG_LOCATION_DIR + CONIIG_FILE, "w") as stream:
+        with open(CONFIG_LOCATION_DIR + CONFIG_FILE, "w") as stream:
             yaml.dump(self.__dict__, stream)
         return self
 
     def load(self):
         """
         Load the configuration from the configuration file
         """
-        with open(CONFIG_LOCATION_DIR + CONIIG_FILE, "r") as stream:
+        with open(CONFIG_LOCATION_DIR + CONFIG_FILE, "r") as stream:
             self.__dict__.update(yaml.load(stream, Loader=yaml.FullLoader))
         return self
 
     def ready(self):
         """
         Check if the configuration is ready to be used
         """
-        return all([self.api_endpoint, self.api_token, self.token_type, self.username])
+        return all([self.api_endpoint, self.api_token, self.token_type, self.username])

labctl/main.py

@@ -13,6 +13,9 @@
 app.add_typer(commands.config_app, name="config", help="Manage the configuration")
 app.add_typer(commands.devices_app, name="devices", help="Manage vpn devices")
 
+if Config().admin_cli:
+    app.add_typer(commands.admin_app, name="admin", help="Admin commands")
+
 @app.callback()
 def callback():
     """
@@ -46,6 +49,7 @@ def me(
     tree.add("[bold]Username:[/bold] " + data.get("username"))
     tree.add("[bold]Email:[/bold] " + data.get("email"))
 
+    # Todo rework this deprecated code
     devices_tree = tree.add(":open_file_folder: Devices")
     for device in data.get('devices_access', []):
         device_tree = devices_tree.add(":computer: " + device.get('name', ':question: Unnamed Device'))