feat: update userctl for zfs

modules/userctl.py

@@ -3,7 +3,7 @@
 import subprocess
 from collections.abc import Sequence
 from datetime import datetime
-from typing import Annotated, Optional
+from typing import Annotated, Mapping, Optional
 
 import typer
 
@@ -19,8 +19,13 @@ def app_callback():
         raise typer.Abort()
 
 
-def run_cmd(cmd: Sequence[str], **kwargs) -> str:
-    return subprocess.check_output(cmd, **kwargs).decode("utf-8").rstrip("\n")
+def run_cmd(cmd: Sequence[str], input: str | None = None) -> str:
+    cmd_kwargs = {}
+
+    if input:
+        cmd_kwargs["input"] = input.encode("utf-8") + b"\n"
+
+    return subprocess.check_output(cmd, **cmd_kwargs).decode("utf-8").rstrip("\n")
 
 
 def check_email(user: str):
@@ -29,7 +34,16 @@ def check_email(user: str):
             "Please provide the university email address, the script determines the username automatically"
         )
 
-    return user.split("@")[0]
+    return user.lower().split("@")[0]
+
+
+def zfs_options(kwargs: Mapping[str, str]) -> list[str]:
+    args: list[str] = []
+
+    for key, value in kwargs.items():
+        args += ["-o", f"{key}={value}"]
+
+    return args
 
 
 @app.command()
@@ -38,40 +52,51 @@ def add(
         str,
         typer.Argument(callback=check_email),
     ],
-    full_name: Annotated[
-        str,
-        typer.Option(),
-    ],
+    full_name: str,
     expire_date: Annotated[
         Optional[datetime],
         typer.Option(formats=[DATE_FORMAT]),
     ] = None,
+    quota: Optional[str] = None,
 ) -> None:
     password = secrets.token_urlsafe()
+    homedir = f"/home/{user}"
+
+    zfs_args: dict[str, str] = {}
+
+    if quota:
+        zfs_args["quota"] = quota
+
+    # https://manpages.ubuntu.com/manpages/jammy/en/man8/zfs.8.html
+    run_cmd(["zfs", "create", *zfs_options(zfs_args), f"data/{homedir}"])
+
+    # copy skeleton
+    run_cmd(["cp", "--recursive", "/etc/skel/.", homedir])
 
-    args: list[str] = [
-        "--create-home",
-        "--base-dir",
-        "/home",
-        "--gid",
-        "100",
+    useradd_args: list[str] = [
+        "--home-dir",
+        homedir,
         "--shell",
         "/bin/bash",
         "--comment",
         full_name,
     ]
 
     if expire_date:
-        args += ["--expiredate", expire_date.strftime(DATE_FORMAT)]
+        useradd_args += ["--expiredate", expire_date.strftime(DATE_FORMAT)]
 
     # https://manpages.ubuntu.com/manpages/jammy/en/man8/useradd.8.html
-    run_cmd(["useradd", *args, user])
+    run_cmd(["useradd", *useradd_args, user])
+
+    # set permissions
+    run_cmd(["chown", "-R", f"{user}:{user}", homedir])
+    run_cmd(["chmod", "750", homedir])
+
+    # https://manpages.ubuntu.com/manpages/jammy/en/man8/chpasswd.8.html
+    run_cmd(["chpasswd"], input=f"{user}:{password}")
 
     # https://manpages.ubuntu.com/manpages/jammy/en/man1/passwd.1.html
-    run_cmd(
-        ["passwd", "--expire", user],
-        input=f"{password}\n{password}\n".encode("utf-8"),
-    )
+    run_cmd(["passwd", "--expire", user])
 
     typer.echo("Login data for new user:")
     typer.echo(f"Username: {user}")
@@ -90,16 +115,16 @@ def remove(
 ) -> None:
     typer.confirm(f"Remove user {user}?", abort=True)
 
-    args: list[str] = []
+    userdel_args: list[str] = []
 
     if force:
-        args.append("--force")
-
-    if not keep_home:
-        args.append("--remove")
+        userdel_args.append("--force")
 
     # https://manpages.ubuntu.com/manpages/jammy/en/man8/userdel.8.html
-    run_cmd(["userdel", *args, user])
+    run_cmd(["userdel", *userdel_args, user])
+
+    if not keep_home:
+        run_cmd(["zfs", "destroy", "--force", f"data/home/{user}"])
 
 
 @app.command()
@@ -112,14 +137,24 @@ def edit(
         Optional[datetime],
         typer.Option(formats=[DATE_FORMAT]),
     ] = None,
+    quota: Optional[str] = None,
 ) -> None:
-    args: list[str] = []
+    usermod_args: list[str] = []
 
     if expire_date:
-        args += ["--expiredate", expire_date.strftime(DATE_FORMAT)]
+        usermod_args += ["--expiredate", expire_date.strftime(DATE_FORMAT)]
 
-    # https://manpages.ubuntu.com/manpages/jammy/en/man8/useradd.8.html
-    run_cmd(["usermod", *args, user])
+    if usermod_args:
+        # https://manpages.ubuntu.com/manpages/jammy/en/man8/useradd.8.html
+        run_cmd(["usermod", *usermod_args, user])
+
+    zfs_args: dict[str, str] = {}
+
+    if quota:
+        zfs_args["quota"] = quota
+
+    if zfs_args:
+        run_cmd(["zfs", "set", *zfs_options(zfs_args), f"data/home/{user}"])
 
 
 if __name__ == "__main__":