From 0f8ed3fc3ae8ef3a649946d0f20b76aeb142cdf4 Mon Sep 17 00:00:00 2001 From: Amir Husayn Panahifar Date: Sat, 12 Sep 2026 13:47:31 +0330 Subject: [PATCH] feat(export): add csv, ansible and permission exports --- src/sshctl/export/__init__.py | 0 src/sshctl/export/exporter.py | 40 ++++++++++++ src/sshctl/export/inventory.py | 108 +++++++++++++++++++++++++++++++ src/sshctl/export/permissions.py | 60 +++++++++++++++++ 4 files changed, 208 insertions(+) create mode 100644 src/sshctl/export/__init__.py create mode 100644 src/sshctl/export/exporter.py create mode 100644 src/sshctl/export/inventory.py create mode 100644 src/sshctl/export/permissions.py diff --git a/src/sshctl/export/__init__.py b/src/sshctl/export/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/sshctl/export/exporter.py b/src/sshctl/export/exporter.py new file mode 100644 index 0000000..0477f4a --- /dev/null +++ b/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 diff --git a/src/sshctl/export/inventory.py b/src/sshctl/export/inventory.py new file mode 100644 index 0000000..4a3e6fa --- /dev/null +++ b/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 diff --git a/src/sshctl/export/permissions.py b/src/sshctl/export/permissions.py new file mode 100644 index 0000000..aa73de9 --- /dev/null +++ b/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