Support ssh configs

src/jinjaturtle/cli.py

@@ -25,7 +25,7 @@ def _build_arg_parser() -> argparse.ArgumentParser:
         "config",
         help=(
             "Path to a config file OR a folder containing supported config files. "
-            "Supported: .toml, .yaml/.yml, .json, .ini/.cfg/.conf, .xml"
+            "Supported: .toml, .yaml/.yml, .json, .ini/.cfg/.conf, .xml, ssh_config/sshd_config"
         ),
     )
     ap.add_argument(
@@ -42,7 +42,7 @@ def _build_arg_parser() -> argparse.ArgumentParser:
     ap.add_argument(
         "-f",
         "--format",
-        choices=["ini", "json", "toml", "yaml", "xml", "postfix", "systemd"],
+        choices=["ini", "json", "toml", "yaml", "xml", "postfix", "systemd", "ssh"],
         help="Force config format instead of auto-detecting from filename.",
     )
     ap.add_argument(

src/jinjaturtle/core.py

@@ -17,6 +17,7 @@ from .handlers import (
     XmlHandler,
     PostfixMainHandler,
     SystemdUnitHandler,
+    SshConfigHandler,
 )
 
 
@@ -61,6 +62,7 @@ _XML_HANDLER = XmlHandler()
 
 _POSTFIX_HANDLER = PostfixMainHandler()
 _SYSTEMD_HANDLER = SystemdUnitHandler()
+_SSH_HANDLER = SshConfigHandler()
 
 _HANDLERS["ini"] = _INI_HANDLER
 _HANDLERS["json"] = _JSON_HANDLER
@@ -70,6 +72,7 @@ _HANDLERS["xml"] = _XML_HANDLER
 
 _HANDLERS["postfix"] = _POSTFIX_HANDLER
 _HANDLERS["systemd"] = _SYSTEMD_HANDLER
+_HANDLERS["ssh"] = _SSH_HANDLER
 
 
 def dump_yaml(data: Any, *, sort_keys: bool = True) -> str:
@@ -132,6 +135,72 @@ def _looks_like_systemd(text: str) -> bool:
     return False
 
 
+def _looks_like_ssh_config(text: str) -> bool:
+    """Conservatively sniff OpenSSH config snippets.
+
+    This is intentionally stricter than generic key/value detection so random
+    .conf files are not misclassified. Exact ssh_config/sshd_config filenames
+    are handled separately above.
+    """
+    meaningful: list[str] = []
+    for line in text.splitlines():
+        stripped = line.strip()
+        if not stripped or stripped.startswith("#"):
+            continue
+        meaningful.append(stripped)
+        if len(meaningful) >= 20:
+            break
+
+    if not meaningful:
+        return False
+
+    ssh_keywords = {
+        "acceptenv",
+        "addressfamily",
+        "allowgroups",
+        "allowtcpforwarding",
+        "allowusers",
+        "authenticationmethods",
+        "authorizedkeysfile",
+        "banner",
+        "ciphers",
+        "chrootdirectory",
+        "denyusers",
+        "forcecommand",
+        "forwardagent",
+        "host",
+        "hostbasedauthentication",
+        "hostkey",
+        "hostname",
+        "identityfile",
+        "include",
+        "kexalgorithms",
+        "listenaddress",
+        "loglevel",
+        "match",
+        "passwordauthentication",
+        "permitrootlogin",
+        "port",
+        "proxycommand",
+        "proxyjump",
+        "pubkeyauthentication",
+        "sendenv",
+        "subsystem",
+        "user",
+        "x11forwarding",
+    }
+
+    hits = 0
+    for line in meaningful:
+        m = re.match(r"^([^\s=#]+)", line)
+        if not m:
+            continue
+        if m.group(1).lower() in ssh_keywords:
+            hits += 1
+
+    return hits >= 2 or (len(meaningful) <= 3 and hits >= 1)
+
+
 def detect_format(path: Path, explicit: str | None = None) -> str:
     """
     Determine config format.
@@ -168,6 +237,8 @@ def detect_format(path: Path, explicit: str | None = None) -> str:
     # well-known filenames
     if name == "main.cf":
         return "postfix"
+    if name in {"ssh_config", "sshd_config"}:
+        return "ssh"
 
     head = _read_head(path)
 
@@ -179,6 +250,8 @@ def detect_format(path: Path, explicit: str | None = None) -> str:
     if suffix in {".conf", ".cf"}:
         if name == "main.cf":
             return "postfix"
+        if _looks_like_ssh_config(head):
+            return "ssh"
         return "ini"
 
     # Fallback: treat as INI-ish

src/jinjaturtle/handlers/__init__.py

@@ -10,6 +10,7 @@ from .xml import XmlHandler
 
 from .postfix import PostfixMainHandler
 from .systemd import SystemdUnitHandler
+from .ssh import SshConfigHandler
 
 __all__ = [
     "BaseHandler",
@@ -21,4 +22,5 @@ __all__ = [
     "XmlHandler",
     "PostfixMainHandler",
     "SystemdUnitHandler",
+    "SshConfigHandler",
 ]

src/jinjaturtle/handlers/ssh.py

@@ -0,0 +1,278 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+import re
+from pathlib import Path
+from typing import Any
+
+from . import BaseHandler
+
+
+_SECTION_KEYWORDS = {"host", "match"}
+
+
+@dataclass
+class SshConfigLine:
+    kind: str  # 'blank' | 'comment' | 'kv' | 'raw'
+    raw: str
+    lineno: int
+    key: str | None = None
+    value: str | None = None
+    processed_value: str | None = None
+    comment: str = ""
+    whitespace_before_comment: str = ""
+    before_value: str = ""
+    newline: str = ""
+    quoted: bool = False
+    is_section: bool = False
+    section_kind: str | None = None
+    section_label: str | None = None
+    context_kind: str | None = None
+    context_label: str | None = None
+    occ_index: int | None = None
+
+
+@dataclass
+class SshConfig:
+    lines: list[SshConfigLine]
+
+
+class SshConfigHandler(BaseHandler):
+    """
+    Handler for OpenSSH ssh_config/sshd_config-style files.
+
+    The format is intentionally treated as keyword + raw argument string rather
+    than attempting to understand the argument type for every OpenSSH keyword.
+    This preserves directives whose values are naturally whitespace-separated,
+    comma-separated, colon-separated, commands, token strings, or quoted strings.
+
+    Host and Match are treated as section headers for variable naming while
+    still templating their criteria like ordinary keyword/argument lines.
+    """
+
+    fmt = "ssh"
+
+    def parse(self, path: Path) -> SshConfig:
+        text = path.read_text(encoding="utf-8")
+        return self._parse_text(text)
+
+    def _parse_text(self, text: str) -> SshConfig:
+        lines = text.splitlines(keepends=True)
+        out: list[SshConfigLine] = []
+        current_kind: str | None = None
+        current_label: str | None = None
+        for lineno, raw_line in enumerate(lines, start=1):
+            content, newline = self._split_newline(raw_line)
+            stripped = content.strip()
+
+            if not stripped:
+                out.append(
+                    SshConfigLine(
+                        kind="blank", raw=raw_line, lineno=lineno, newline=newline
+                    )
+                )
+                continue
+
+            if content.lstrip(" \t").startswith("#"):
+                out.append(
+                    SshConfigLine(
+                        kind="comment", raw=raw_line, lineno=lineno, newline=newline
+                    )
+                )
+                continue
+
+            parsed = self._split_keyword_value(content)
+            if parsed is None:
+                out.append(
+                    SshConfigLine(
+                        kind="raw", raw=raw_line, lineno=lineno, newline=newline
+                    )
+                )
+                continue
+
+            key, before_value, value_and_comment = parsed
+            value_part, comment = self._split_inline_comment(value_and_comment, {"#"})
+            whitespace_before_comment = value_part[len(value_part.rstrip(" \t")) :]
+            raw_value = value_part.strip()
+
+            quoted = (
+                len(raw_value) >= 2
+                and raw_value[0] == raw_value[-1]
+                and raw_value[0] in {'"', "'"}
+            )
+            processed_value = raw_value[1:-1] if quoted else raw_value
+
+            key_lower = key.lower()
+            is_section = key_lower in _SECTION_KEYWORDS
+            section_kind = key if is_section else None
+            section_label = None
+            context_kind = current_kind
+            context_label = current_label
+
+            if is_section:
+                # Host/Match lines define the context for subsequent lines, but
+                # the line itself is kept globally named as the section marker.
+                base_label = self._section_label(processed_value, lineno)
+                section_label = base_label
+                current_kind = key
+                current_label = base_label
+                context_kind = None
+                context_label = None
+
+            out.append(
+                SshConfigLine(
+                    kind="kv",
+                    raw=raw_line,
+                    lineno=lineno,
+                    key=key,
+                    value=raw_value,
+                    processed_value=processed_value,
+                    comment=comment,
+                    whitespace_before_comment=whitespace_before_comment,
+                    before_value=before_value,
+                    newline=newline,
+                    quoted=quoted,
+                    is_section=is_section,
+                    section_kind=section_kind,
+                    section_label=section_label,
+                    context_kind=context_kind,
+                    context_label=context_label,
+                )
+            )
+
+        unit = SshConfig(lines=out)
+        self._assign_occurrences(unit)
+        return unit
+
+    @staticmethod
+    def _split_newline(raw_line: str) -> tuple[str, str]:
+        if raw_line.endswith("\r\n"):
+            return raw_line[:-2], "\r\n"
+        if raw_line.endswith("\n"):
+            return raw_line[:-1], "\n"
+        return raw_line, ""
+
+    @staticmethod
+    def _split_keyword_value(content: str) -> tuple[str, str, str] | None:
+        """Return (key, text_before_value, value_and_comment)."""
+        m = re.match(r"^([ \t]*)([^\s=#]+)(.*)$", content)
+        if not m:
+            return None
+
+        leading, key, rest = m.groups()
+        # Keep the original spelling/indentation for rendering, but allow either:
+        #   Keyword value
+        #   Keyword=value
+        #   Keyword = value
+        i = 0
+        while i < len(rest) and rest[i] in " \t":
+            i += 1
+        if i < len(rest) and rest[i] == "=":
+            i += 1
+            while i < len(rest) and rest[i] in " \t":
+                i += 1
+        elif i == 0 and rest:
+            # Some unexpected non-whitespace separator. Treat as raw.
+            return None
+
+        before_value = leading + key + rest[:i]
+        value_and_comment = rest[i:]
+        return key, before_value, value_and_comment
+
+    def _section_label(self, value: str | None, lineno: int) -> str:
+        raw = (value or "").strip()
+        if not raw:
+            return f"line_{lineno}"
+        # Use the same sanitisation semantics as variable names, but without a
+        # role prefix. This keeps labels such as '*' useful instead of empty.
+        label = self.make_var_name("x", (raw,))
+        if label.startswith("x_"):
+            label = label[2:]
+        elif label == "x":
+            label = "all"
+        label = re.sub(r"_+", "_", label).strip("_")
+        return label or f"line_{lineno}"
+
+    def _base_path(self, ln: SshConfigLine) -> tuple[str, ...]:
+        if ln.kind != "kv" or not ln.key:
+            return ()
+        if ln.is_section and ln.section_kind and ln.section_label:
+            return (ln.section_kind, ln.section_label)
+        if ln.context_kind and ln.context_label:
+            return (ln.context_kind, ln.context_label, ln.key)
+        return (ln.key,)
+
+    def _assign_occurrences(self, parsed: SshConfig) -> None:
+        counts: dict[tuple[str, ...], int] = {}
+        for ln in parsed.lines:
+            if ln.kind == "kv":
+                base = self._base_path(ln)
+                if base:
+                    counts[base] = counts.get(base, 0) + 1
+
+        seen: dict[tuple[str, ...], int] = {}
+        for ln in parsed.lines:
+            if ln.kind != "kv":
+                continue
+            base = self._base_path(ln)
+            if not base:
+                continue
+            idx = seen.get(base, 0)
+            seen[base] = idx + 1
+            if counts.get(base, 0) > 1:
+                ln.occ_index = idx
+
+    def _path_for_line(self, ln: SshConfigLine) -> tuple[str, ...]:
+        base = self._base_path(ln)
+        if ln.occ_index is not None:
+            return base + (str(ln.occ_index),)
+        return base
+
+    def flatten(self, parsed: Any) -> list[tuple[tuple[str, ...], Any]]:
+        if not isinstance(parsed, SshConfig):
+            raise TypeError("SSH config parse result must be an SshConfig")
+
+        items: list[tuple[tuple[str, ...], Any]] = []
+        for ln in parsed.lines:
+            if ln.kind != "kv":
+                continue
+            path = self._path_for_line(ln)
+            if not path:
+                continue
+            items.append((path, ln.processed_value or ""))
+        return items
+
+    def generate_jinja2_template(
+        self,
+        parsed: Any,
+        role_prefix: str,
+        original_text: str | None = None,
+    ) -> str:
+        if not isinstance(parsed, SshConfig):
+            raise TypeError("SSH config parse result must be an SshConfig")
+
+        out_lines: list[str] = []
+        for ln in parsed.lines:
+            if ln.kind != "kv":
+                out_lines.append(ln.raw)
+                continue
+
+            path = self._path_for_line(ln)
+            if not path:
+                out_lines.append(ln.raw)
+                continue
+
+            var = self.make_var_name(role_prefix, path)
+            if ln.quoted and ln.value:
+                quote_char = ln.value[0]
+                replacement_value = f"{quote_char}{{{{ {var} }}}}{quote_char}"
+            else:
+                replacement_value = f"{{{{ {var} }}}}"
+
+            rendered = (
+                f"{ln.before_value}{replacement_value}"
+                f"{ln.whitespace_before_comment}{ln.comment}{ln.newline}"
+            )
+            out_lines.append(rendered)
+
+        return "".join(out_lines)