Merge pull request 'hidden-ssh' (#21) from hidden-ssh into main

Reviewed-on: clan/clan-core#21

.gitignore

@@ -7,4 +7,6 @@ __pycache__
 .mypy_cache
 .pytest_cache
 .pythonenv
+.reports
 .ruff_cache
+htmlcov

flake-parts/packages.nix

@@ -1,11 +1,17 @@
 { self, lib, ... }: {
-  flake.packages.x86_64-linux = {
-    install-iso = (lib.nixosSystem {
-      system = "x86_64-linux";
-      modules = [
-        self.nixosModules.installer
-        self.inputs.nixos-generators.nixosModules.all-formats
-      ];
-    }).config.formats.install-iso;
-  };
+  flake.packages.x86_64-linux =
+    let
+      installer = lib.nixosSystem {
+        system = "x86_64-linux";
+        modules = [
+          self.nixosModules.installer
+          self.inputs.nixos-generators.nixosModules.all-formats
+        ];
+      };
+    in
+    {
+      install-iso = installer.config.formats.install-iso;
+      install-vm-nogui = installer.config.formats.vm-nogui;
+      install-vm = installer.config.formats.vm;
+    };
 }

flake.nix

@@ -31,12 +31,12 @@
           installer = {
             imports = [
               ./installer.nix
-              ./hidden-announce.nix
+              ./hidden-ssh-announce.nix
             ];
           };
           hidden-announce = {
             imports = [
-              ./hidden-announce.nix
+              ./hidden-ssh-announce.nix
             ];
           };
         };

hidden-ssh-announce.nix

@@ -3,11 +3,11 @@
 , pkgs
 , ...
 }: {
-  options.hidden-announce = {
-    enable = lib.mkEnableOption "hidden-announce";
+  options.hidden-ssh-announce = {
+    enable = lib.mkEnableOption "hidden-ssh-announce";
     script = lib.mkOption {
       type = lib.types.package;
-      default = pkgs.writers.writeDash "test-output";
+      default = pkgs.writers.writeDash "test-output" "echo $1";
       description = ''
         script to run when the hidden tor service was started and they hostname is known.
         takes the hostname as $1
@@ -15,7 +15,8 @@
     };
   };
 
-  config = lib.mkIf config.hidden-announce.enable {
+  config = lib.mkIf config.hidden-ssh-announce.enable {
+    services.openssh.enable = true;
     services.tor = {
       enable = true;
       relay.onionServices.hidden-ssh = {
@@ -43,7 +44,7 @@
             sleep 1
           done
 
-          ${config.hidden-announce.script} "$(cat ${config.services.tor.settings.DataDirectory}/onion/hidden-ssh/hostname)"
+          ${config.hidden-ssh-announce.script} "$(cat ${config.services.tor.settings.DataDirectory}/onion/hidden-ssh/hostname)"
         '';
         PrivateTmp = "true";
         User = "tor";

installer.nix

@@ -7,19 +7,31 @@
   ];
   services.openssh.settings.PermitRootLogin = "yes";
   system.activationScripts.root-password = ''
+    mkdir -p /var/shared
     ${pkgs.pwgen}/bin/pwgen -s 16 1 > /var/shared/root-password
     echo "root:$(cat /var/shared/root-password)" | chpasswd
   '';
-  hidden-announce = {
+  hidden-ssh-announce = {
     enable = true;
     script = pkgs.writers.writeDash "write-hostname" ''
+      mkdir -p /var/shared
       echo "$1" > /var/shared/onion-hostname
+      ${pkgs.jq}/bin/jq -nc \
+        --arg password "$(cat /var/shared/root-password)" \
+        --arg address "$(cat /var/shared/onion-hostname)" '{
+          password: $password, address: $address
+        }' > /var/shared/login.info
+      cat /var/shared/login.info |
+        ${pkgs.qrencode}/bin/qrencode -t utf8 > /var/shared/qrcode.utf8
+      cat /var/shared/login.info |
+        ${pkgs.qrencode}/bin/qrencode -t png > /var/shared/qrcode.png
     '';
   };
   services.getty.autologinUser = lib.mkForce "root";
   programs.bash.interactiveShellInit = ''
     if [ "$(tty)" = "/dev/tty1" ]; then
-      echo "ssh://root:$(cat /var/shared/root-password)@$(cat /var/shared/onion-hostname)"
+      until test -e /var/shared/qrcode.utf8; do sleep 1; done
+      cat /var/shared/qrcode.utf8
     fi
   '';
   formatConfigs.install-iso = {

pkgs/clan-cli/clan_cli/__init__.py

@@ -2,7 +2,7 @@
 import argparse
 import sys
 
-from . import admin
+from . import admin, ssh
 
 has_argcomplete = True
 try:
@@ -18,12 +18,20 @@ def main() -> None:
 
     parser_admin = subparsers.add_parser("admin")
     admin.register_parser(parser_admin)
+
+    parser_ssh = subparsers.add_parser("ssh", help="ssh to a remote machine")
+    ssh.register_parser(parser_ssh)
+
     if has_argcomplete:
         argcomplete.autocomplete(parser)
-    parser.parse_args()
+
     if len(sys.argv) == 1:
         parser.print_help()
 
+    args = parser.parse_args()
+    if hasattr(args, "func"):
+        args.func(args)  # pragma: no cover
+
 
 if __name__ == "__main__":  # pragma: no cover
     main()

pkgs/clan-cli/clan_cli/ssh.py

@@ -0,0 +1,84 @@
+import argparse
+import json
+import subprocess
+from typing import Optional
+
+
+def ssh(
+    host: str,
+    user: str = "root",
+    password: Optional[str] = None,
+    ssh_args: list[str] = [],
+) -> None:
+    nix_shell_args = []
+    password_args = []
+    if password:
+        nix_shell_args = [
+            "nix",
+            "shell",
+            "nixpkgs#sshpass",
+            "-c",
+        ]
+        password_args = [
+            "sshpass",
+            "-p",
+            password,
+        ]
+    _ssh_args = ssh_args + [
+        "ssh",
+        "-o",
+        "UserKnownHostsFile=/dev/null",
+        "-o",
+        "StrictHostKeyChecking=no",
+        f"{user}@{host}",
+    ]
+    cmd = nix_shell_args + ["torify"] + password_args + _ssh_args
+    subprocess.run(cmd)
+
+
+def qrcode_scan(pictureFile: str) -> str:
+    return (
+        subprocess.run(
+            [
+                "nix",
+                "shell",
+                "nixpkgs#zbar",
+                "-c",
+                "zbarimg",
+                "--quiet",
+                "--raw",
+                pictureFile,
+            ],
+            stdout=subprocess.PIPE,
+            check=True,
+        )
+        .stdout.decode()
+        .strip()
+    )
+
+
+def main(args: argparse.Namespace) -> None:  # pragma: no cover
+    if args.json:
+        with open(args.json) as file:
+            ssh_data = json.load(file)
+        ssh(host=ssh_data["address"], password=ssh_data["password"])
+    elif args.png:
+        ssh_data = json.loads(qrcode_scan(args.png))
+        ssh(host=ssh_data["address"], password=ssh_data["password"])
+
+
+def register_parser(parser: argparse.ArgumentParser) -> None:
+    group = parser.add_mutually_exclusive_group(required=True)
+    group.add_argument(
+        "-j",
+        "--json",
+        help="specify the json file for ssh data (generated by starting the clan installer)",
+    )
+    group.add_argument(
+        "-P",
+        "--png",
+        help="specify the json file for ssh data as the qrcode image (generated by starting the clan installer)",
+    )
+    # TODO pass all args we don't parse into ssh_args, currently it fails if arg starts with -
+    parser.add_argument("ssh_args", nargs="*", default=[])
+    parser.set_defaults(func=main)

pkgs/clan-cli/default.nix

@@ -41,7 +41,7 @@ let
     propagatedBuildInputs =
       dependencies
       ++ [ ];
-    passthru.tests = { inherit clan-tests clan-mypy; };
+    passthru.tests = { inherit clan-black clan-mypy clan-pytest clan-ruff; };
     passthru.devDependencies = devDependencies;
     postInstall = ''
       installShellCompletion --bash --name clan \
@@ -49,10 +49,19 @@ let
       installShellCompletion --fish --name clan.fish \
         <(${python3.pkgs.argcomplete}/bin/register-python-argcomplete --shell fish clan)
     '';
+    meta.mainProgram = "clan";
   };
 
   checkPython = python3.withPackages (_ps: devDependencies ++ dependencies);
 
+  clan-black = runCommand "${name}-black" { } ''
+    cp -r ${src} ./src
+    chmod +w -R ./src
+    cd src
+    ${checkPython}/bin/black --check .
+    touch $out
+  '';
+
   clan-mypy = runCommand "${name}-mypy" { } ''
     cp -r ${src} ./src
     chmod +w -R ./src
@@ -61,11 +70,20 @@ let
     touch $out
   '';
 
-  clan-tests = runCommand "${name}-tests" { } ''
+  clan-pytest = runCommand "${name}-tests" { } ''
     cp -r ${src} ./src
     chmod +w -R ./src
     cd src
-    ${checkPython}/bin/python -m pytest ./tests
+    ${checkPython}/bin/python -m pytest ./tests \
+      || echo -e "generate coverage report py running:\n  pytest; firefox .reports/html/index.html"
+    touch $out
+  '';
+
+  clan-ruff = runCommand "${name}-ruff" { } ''
+    cp -r ${src} ./src
+    chmod +w -R ./src
+    cd src
+    ${pkgs.ruff}/bin/ruff check .
     touch $out
   '';
 

pkgs/clan-cli/flake-module.nix

@@ -7,6 +7,7 @@
     in
     {
       packages.${name} = package;
+      packages.default = package;
       checks = package.tests;
     };
 }

pkgs/clan-cli/pyproject.toml

@@ -9,7 +9,7 @@ dynamic = ["version"]
 scripts = {clan = "clan_cli:main"}
 
 [tool.pytest.ini_options]
-addopts = "--cov . --cov-report term --cov-fail-under=100 --no-cov-on-fail"
+addopts = "--cov . --cov-report term --cov-report html:.reports/html --cov-fail-under=100 --no-cov-on-fail"
 
 [tool.mypy]
 python_version = "3.10"

pkgs/clan-cli/tests/test_clan_ssh.py

@@ -0,0 +1,66 @@
+import sys
+from typing import Union
+
+import pytest
+import pytest_subprocess.fake_process
+from pytest_subprocess import utils
+
+import clan_cli.ssh
+
+
+def test_no_args(
+    capsys: pytest.CaptureFixture, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    monkeypatch.setattr(sys, "argv", ["", "ssh"])
+    with pytest.raises(SystemExit):
+        clan_cli.main()
+    captured = capsys.readouterr()
+    assert captured.err.startswith("usage:")
+
+
+# using fp fixture from pytest-subprocess
+def test_ssh_no_pass(fp: pytest_subprocess.fake_process.FakeProcess) -> None:
+    host = "somehost"
+    user = "user"
+    cmd: list[Union[str, utils.Any]] = [
+        "torify",
+        "ssh",
+        "-o",
+        "UserKnownHostsFile=/dev/null",
+        "-o",
+        "StrictHostKeyChecking=no",
+        f"{user}@{host}",
+        fp.any(),
+    ]
+    fp.register(cmd)
+    clan_cli.ssh.ssh(
+        host=host,
+        user=user,
+    )
+    assert fp.call_count(cmd) == 1
+
+
+def test_ssh_with_pass(fp: pytest_subprocess.fake_process.FakeProcess) -> None:
+    host = "somehost"
+    user = "user"
+    cmd: list[Union[str, utils.Any]] = [
+        "nix",
+        "shell",
+        "nixpkgs#sshpass",
+        "-c",
+        fp.any(),
+    ]
+    fp.register(cmd)
+    clan_cli.ssh.ssh(
+        host=host,
+        user=user,
+        password="XXX",
+    )
+    assert fp.call_count(cmd) == 1
+
+
+def test_qrcode_scan(fp: pytest_subprocess.fake_process.FakeProcess) -> None:
+    cmd: list[Union[str, utils.Any]] = [fp.any()]
+    fp.register(cmd, stdout="https://test.test")
+    result = clan_cli.ssh.qrcode_scan("test.png")
+    assert result == "https://test.test"