1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
|
"""Install and toggle a secondary Canonical VPN without tracking credentials."""
import argparse
import os
import shlex
import subprocess
import tempfile
import zipfile
from pathlib import Path
from scripts.canonical import require_canonical
NAME = "canonical-secondary"
REFERENCES = {"ca", "cert", "key", "tls-auth"}
def output(*args):
return subprocess.check_output(["nmcli", *args], text=True).strip()
def run(command):
subprocess.run(command, check=True)
def secondary_files(archive, endpoint, destination):
with zipfile.ZipFile(archive) as source:
names = source.namelist()
candidates = [
name
for name in names
if name.startswith(f"{endpoint}-")
and name.endswith("@2.conf")
and "/" not in name
]
if len(candidates) != 1 or len(names) != len(set(names)):
raise ValueError(
"Archive must contain one secondary profile for the selected endpoint."
)
lines = source.read(candidates[0]).decode().splitlines()
files, rendered, found = {}, [], set()
for line in lines:
words = shlex.split(line, comments=True)
if words and words[0] in REFERENCES:
directive, filename = words[:2]
if (
directive in found
or Path(filename).name != filename
or filename in (".", "..")
):
raise ValueError(
"Credential references must be unique plain filenames."
)
if directive in ("cert", "key") and not filename.endswith(
f"@2.{'crt' if directive == 'cert' else 'key'}"
):
raise ValueError(
"Refusing credentials that are not for the secondary identity."
)
files[filename] = source.read(filename)
path = (
str(destination / filename)
.replace("\\", "\\\\")
.replace('"', '\\"')
)
line = f'{directive} "{path}"' + (
" " + " ".join(words[2:]) if words[2:] else ""
)
found.add(directive)
rendered.append(line)
if found != REFERENCES:
raise ValueError(
"Profile must reference CA, secondary certificate/key, and TLS auth key."
)
files[f"{NAME}.conf"] = ("\n".join(rendered) + "\n").encode()
return files
def write_credentials(destination, files):
if destination.is_symlink() or destination.parent.is_symlink():
raise ValueError("Credential directories must not be symlinks.")
destination.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
if destination.exists():
if any(
(destination / name).is_symlink()
or not (destination / name).is_file()
or (destination / name).read_bytes() != data
for name, data in files.items()
):
raise ValueError("Existing credentials differ; no files were replaced.")
destination.chmod(0o700)
for name in files:
(destination / name).chmod(0o600)
return
with tempfile.TemporaryDirectory(
prefix=".vpn-", dir=destination.parent
) as temporary:
staging = Path(temporary) / "credentials"
staging.mkdir(mode=0o700)
for name, data in files.items():
path = staging / name
path.write_bytes(data)
path.chmod(0o600)
staging.rename(destination)
def routing(mode):
if mode not in ("full", "split"):
raise ValueError("Routing mode must be full or split.")
settings = []
for family in ("ipv4", "ipv6"):
settings.extend(
[
f"{family}.never-default",
"no" if mode == "full" else "yes",
f"{family}.dns-search",
"~." if mode == "full" else "",
f"{family}.dns-priority",
"-50" if mode == "full" else "50",
]
)
return settings
def install(archive, endpoint):
if NAME in output("-g", "NAME", "connection", "show").splitlines():
raise ValueError(f"Profile {NAME} already exists; it was not replaced.")
destination = Path.home() / ".sesame" / "canonical-secondary"
files = secondary_files(archive, endpoint, destination)
write_credentials(destination, files)
run(
[
"sudo",
"nmcli",
"connection",
"import",
"type",
"openvpn",
"file",
str(destination / f"{NAME}.conf"),
]
)
run(
[
"sudo",
"nmcli",
"connection",
"modify",
"id",
NAME,
"connection.autoconnect",
"no",
"connection.permissions",
f"user:{os.environ['USER']}",
*routing("split"),
]
)
print(f"Installed {NAME}, disconnected, with split routing selected.")
def up(mode):
if NAME in output("-g", "NAME", "connection", "show", "--active").splitlines():
raise ValueError(
"Disconnect the secondary VPN before changing its routing mode."
)
run(["sudo", "nmcli", "connection", "modify", "id", NAME, *routing(mode)])
run(["nmcli", "--ask", "connection", "up", "id", NAME])
def main():
parser = argparse.ArgumentParser(description=__doc__)
commands = parser.add_subparsers(dest="action", required=True)
setup = commands.add_parser("install")
setup.add_argument("archive", type=Path)
setup.add_argument("--endpoint", choices=("uk", "us", "tw"), default="uk")
connect = commands.add_parser("up")
connect.add_argument("mode", choices=("full", "split"), default="split", nargs="?")
commands.add_parser("down")
args = parser.parse_args()
require_canonical()
if args.action == "install":
install(args.archive, args.endpoint)
elif args.action == "up":
up(args.mode)
else:
run(["nmcli", "connection", "down", "id", NAME])
if __name__ == "__main__":
try:
main()
except ValueError as error:
raise SystemExit(str(error)) from None
except (
KeyError,
OSError,
zipfile.BadZipFile,
subprocess.CalledProcessError,
):
raise SystemExit(
"VPN setup failed. Check the archive, installed OpenVPN plug-in, and profile state."
) from None
|
