feat(export): add csv, ansible and permission exports

src/sshctl/export/exporter.py

@@ -0,0 +1,40 @@
+from __future__ import annotations
+
+import csv
+import io
+from collections.abc import Sequence
+from pathlib import Path
+
+from sshctl.config.resolver import ResolvedHost, resolve_all_hosts
+
+
+def export_csv(
+    hosts: Sequence[ResolvedHost] | None = None,
+    *,
+    config_path: str | Path | None = None,
+) -> str:
+    if hosts is None:
+        if config_path is None:
+            raise ValueError("Either hosts or config_path must be provided")
+        hosts = resolve_all_hosts(config_path)
+
+    output = io.StringIO()
+    writer = csv.writer(output)
+    writer.writerow(["Host", "HostName", "User", "Port", "ProxyJump", "IdentityFile"])
+
+    for h in hosts:
+        writer.writerow([h.alias, h.hostname, h.user, h.port, h.proxy_jump, h.identity_file])
+
+    return output.getvalue()
+
+
+def export_csv_to_file(
+    path: str | Path,
+    hosts: Sequence[ResolvedHost] | None = None,
+    *,
+    config_path: str | Path | None = None,
+) -> Path:
+    path = Path(path)
+    content = export_csv(hosts, config_path=config_path)
+    path.write_text(content, encoding="utf-8")
+    return path

src/sshctl/export/inventory.py

@@ -0,0 +1,108 @@
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Any
+
+import yaml
+
+from sshctl.config.parser import resolve_includes
+from sshctl.core.models import SshConfig
+
+
+def _build_ansible_groups(
+    config: SshConfig,
+    base_dir: Path,
+) -> dict[str, Any]:
+    groups: dict[str, Any] = {}
+    conf_d = base_dir / "conf.d"
+
+    if not conf_d.exists():
+        return groups
+
+    _walk_conf_d(conf_d, groups, config)
+    return groups
+
+
+def _walk_conf_d(
+    path: Path,
+    groups: dict[str, Any],
+    config: SshConfig,
+) -> None:
+    children: list[Path] = sorted(
+        [p for p in path.iterdir() if p.is_dir()],
+        key=lambda p: p.name,
+    )
+    conf_files: list[Path] = sorted(
+        [p for p in path.glob("*.conf")],
+        key=lambda p: p.name,
+    )
+
+    group_name = _to_group_name(path.name)
+
+    group: dict[str, Any] = {"hosts": {}}
+
+    for conf_file in conf_files:
+        from sshctl.config.parser import parse_file
+
+        try:
+            parsed = parse_file(conf_file)
+            for block in parsed.blocks:
+                if block.alias == "*":
+                    continue
+                host_vars: dict[str, str] = {}
+                hostname = block.get("HostName")
+                if hostname:
+                    host_vars["ansible_host"] = hostname
+                user = block.get("User")
+                if user:
+                    host_vars["ansible_user"] = user
+                port = block.get("Port")
+                if port:
+                    host_vars["ansible_port"] = port
+                proxy = block.get("ProxyJump")
+                if proxy:
+                    host_vars["ansible_ssh_common_args"] = f"-J {proxy}"
+                group["hosts"][block.alias] = host_vars
+        except Exception:
+            pass
+
+    groups[group_name] = group
+
+    for child in children:
+        _walk_conf_d(child, groups, config)
+
+
+def _to_group_name(name: str) -> str:
+    name = name.replace("-", "_").replace(".", "_")
+    if name[0].isdigit():
+        name = f"z{name}"
+    return name
+
+
+def generate_inventory(ssh_dir: str | Path) -> dict[str, Any]:
+    ssh_dir = Path(ssh_dir).expanduser().resolve()
+    config_path = ssh_dir / "config"
+    if not config_path.exists():
+        return {}
+
+    config = resolve_includes(config_path)
+    return _build_ansible_groups(config, ssh_dir)
+
+
+def write_inventory(
+    ssh_dir: str | Path,
+    output_dir: str | Path | None = None,
+) -> Path:
+    ssh_dir = Path(ssh_dir).expanduser().resolve()
+    inventory = generate_inventory(ssh_dir)
+
+    if output_dir is None:
+        output_dir = ssh_dir / "inventory.d"
+    output_dir = Path(output_dir)
+    output_dir.mkdir(parents=True, exist_ok=True)
+
+    out_path = output_dir / "hosts.yml"
+    yaml_str = yaml.dump(inventory, default_flow_style=False, sort_keys=True)
+    out_path.write_text(yaml_str, encoding="utf-8")
+
+    return out_path

src/sshctl/export/permissions.py

@@ -0,0 +1,60 @@
+from __future__ import annotations
+
+import stat
+from pathlib import Path
+
+from sshctl.core.exceptions import PermissionFixError
+
+
+def fix_ssh_permissions(ssh_dir: str | Path) -> dict[str, int]:
+    ssh_dir = Path(ssh_dir).expanduser().resolve()
+    if not ssh_dir.exists():
+        raise PermissionFixError(f"SSH directory not found: {ssh_dir}")
+
+    counts: dict[str, int] = {"dirs": 0, "private": 0, "public": 0, "skipped": 0}
+
+    for entry in [ssh_dir] + list(ssh_dir.rglob("*")):
+        if entry.name.startswith(".git"):
+            continue
+        try:
+            if entry.is_dir():
+                entry.chmod(stat.S_IRWXU)
+                counts["dirs"] += 1
+            elif entry.suffix == ".pub" or entry.name.endswith("-cert.pub"):
+                entry.chmod(stat.S_IRUSR | stat.S_IWUSR | stat.S_IRGRP | stat.S_IROTH)
+                counts["public"] += 1
+            elif entry.name == ".gitkeep":
+                counts["skipped"] += 1
+            else:
+                entry.chmod(stat.S_IRUSR | stat.S_IWUSR)
+                counts["private"] += 1
+        except PermissionError:
+            counts["skipped"] += 1
+
+    return counts
+
+
+def check_permissions(ssh_dir: str | Path) -> dict[str, list[str]]:
+    ssh_dir = Path(ssh_dir).expanduser().resolve()
+    issues: dict[str, list[str]] = {"bad_dir": [], "bad_private": [], "bad_public": []}
+
+    for entry in ssh_dir.rglob("*"):
+        if entry.name.startswith(".git"):
+            continue
+        try:
+            mode = entry.stat().st_mode
+            if entry.is_dir():
+                if mode & stat.S_IRWXU != stat.S_IRWXU:
+                    issues["bad_dir"].append(str(entry))
+            elif entry.suffix == ".pub":
+                expected = stat.S_IRUSR | stat.S_IWUSR | stat.S_IRGRP | stat.S_IROTH
+                if mode & 0o777 != expected:
+                    issues["bad_public"].append(str(entry))
+            elif entry.name not in (".gitkeep",):
+                expected = stat.S_IRUSR | stat.S_IWUSR
+                if mode & 0o777 != expected:
+                    issues["bad_private"].append(str(entry))
+        except PermissionError:
+            pass
+
+    return issues