authentik.outposts.docker_ssh

 1"""Docker SSH helper"""
 2
 3import os
 4from pathlib import Path
 5from tempfile import gettempdir
 6
 7from docker.errors import DockerException
 8
 9from authentik.crypto.models import CertificateKeyPair
10
11HEADER = "### Managed by authentik"
12FOOTER = "### End Managed by authentik"
13
14
15def opener(path: Path | str, flags: int):
16    """File opener to create files as 600 perms"""
17    return os.open(path, flags, 0o600)
18
19
20class SSHManagedExternallyException(DockerException):
21    """Raised when the ssh config file is managed externally."""
22
23
24class DockerInlineSSH:
25    """Create paramiko ssh config from CertificateKeyPair"""
26
27    host: str
28    keypair: CertificateKeyPair
29
30    key_path: str
31    config_path: Path
32    header: str
33
34    def __init__(self, host: str, keypair: CertificateKeyPair) -> None:
35        self.host = host
36        self.keypair = keypair
37        self.config_path = Path("~/.ssh/config").expanduser()
38        if self.config_path.exists() and HEADER not in self.config_path.read_text(encoding="utf-8"):
39            # SSH Config file already exists and there's no header from us, meaning that it's
40            # been externally mapped into the container for more complex configs
41            raise SSHManagedExternallyException(
42                "SSH Config exists and does not contain authentik header"
43            )
44        if not self.keypair:
45            raise DockerException("keypair must be set for SSH connections")
46        self.header = f"{HEADER} - {self.host}\n"
47
48    def write_config(self, key_path: str) -> bool:
49        """Update the local user's ssh config file"""
50        with open(self.config_path, "a+", encoding="utf-8") as ssh_config:
51            if self.header in ssh_config.readlines():
52                return False
53            ssh_config.writelines(
54                [
55                    self.header,
56                    f"Host {self.host}\n",
57                    f"    IdentityFile {key_path}\n",
58                    "    StrictHostKeyChecking No\n",
59                    "    UserKnownHostsFile /dev/null\n",
60                    f"{FOOTER}\n",
61                    "\n",
62                ]
63            )
64        return True
65
66    def write_key(self):
67        """Write keypair's private key to a temporary file"""
68        path = Path(gettempdir(), f"{self.keypair.pk}_private.pem")
69        with open(path, "w", encoding="utf8", opener=opener) as _file:
70            _file.write(self.keypair.key_data)
71        return str(path)
72
73    def write(self):
74        """Write keyfile and update ssh config"""
75        self.key_path = self.write_key()
76        was_written = self.write_config(self.key_path)
77        if not was_written:
78            self.cleanup()
79
80    def cleanup(self):
81        """Cleanup when we're done"""
82        try:
83            os.unlink(self.key_path)
84            with open(self.config_path, encoding="utf-8") as ssh_config:
85                start = 0
86                end = 0
87                lines = ssh_config.readlines()
88                for idx, line in enumerate(lines):
89                    if line == self.header:
90                        start = idx
91                    if start != 0 and line == f"{FOOTER}\n":
92                        end = idx
93            with open(self.config_path, "w+", encoding="utf-8") as ssh_config:
94                lines = lines[:start] + lines[end + 2 :]
95                ssh_config.writelines(lines)
96        except OSError:
97            # If we fail deleting a file it doesn't matter that much
98            # since we're just in a container
99            pass