~sumner/nixos-configuration

nixos-configuration/archives/initialize-droplet.py -rwxr-xr-x 6.0 KiB
17aa893fSumner Evans matrix-synapse: 1.50.0rc2 -> 1.50.1 2 days ago
                                                                                
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
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
#! /bin/env python3
"""
Environment variables:

    DIGITALOCEAN_ACCESS_TOKEN
        The token to use to authenticate to the DigitalOcean API.
"""

import os
import sys
import subprocess
from pathlib import Path

import digitalocean


def prompt_select(
    prompt,
    options,
    formatter,
    multiple=False,
    default=None,
    allow_none=False,
):
    options = list(options)
    while True:
        print()
        print(prompt)
        default_idx = None
        for i, opt in enumerate(options):
            print(f'  {i}: {formatter(opt)}')
            if default and opt.slug == default:
                default_idx = i
        print()

        try:
            how_many = {
                (True, True): 'zero or more',
                (True, False): 'one or more',
                (False, True): 'zero or one',
                (False, False): 'one',
            }[(multiple, allow_none)]

            result = set(
                map(
                    int,
                    input('Enter {} of the indexes{}: '.format(
                        how_many,
                        f' [{default_idx}]' if default else '',
                    )).split(),
                ))

            if len(result) == 0:
                if default:
                    if multiple:
                        return [options[default_idx]]
                    else:
                        return options[default_idx]
                elif allow_none:
                    return None
                continue
            elif len(result) == 1:
                if multiple:
                    return [options[list(result)[0]]]
                else:
                    return options[list(result)[0]]
            elif multiple is False:
                continue
            else:
                return [options[i] for i in result]
        except Exception:
            # Could be something that wasn't a number, invalid index, etc.
            # Regardless, reprompt.
            pass


def prompt_proceed(prompt):
    while True:
        proceed = input(prompt + ' [yN]: ')
        if proceed in ('y', 'Y'):
            return True
        elif proceed in ('n', 'N', ''):
            return False


token = (os.environ.get('DIGITALOCEAN_ACCESS_TOKEN')
         or input('DigitalOcean Access Token: '))

print(f'Using Access Token: {token}')
manager = digitalocean.Manager(token=token)

ips = manager.get_all_floating_ips()
floating_ip_str = os.environ.get('DROPLET_FLOATING_IP')
floating_ip_to_use = None if not floating_ip_str else [
    i for i in ips if i.ip == floating_ip_str
][0]
if not floating_ip_to_use:
    floating_ip_to_use = prompt_select(
        'Which floating IP do you want to assign to the droplet?',
        ips,
        lambda i: str(i),
        allow_none=True,
    )

# Prompt for the keys to auto-add to the droplet.
keys_to_use = prompt_select(
    'Which SSH keys do you want to be able to access the machine?',
    manager.get_all_sshkeys(),
    lambda s: s.name,
    multiple=True,
)

name = os.environ.get('DROPLET_NAME') or input('Droplet name: ')

region = os.environ.get('DROPLET_REGION')
if not region:
    region = prompt_select(
        'Which droplet region do you want to use?',
        manager.get_all_regions(),
        lambda r: f'{r.name}: {r.slug}',
        default='sfo2',
    ).slug

size = os.environ.get('DROPLET_SIZE')
if not size:
    size = prompt_select(
        'Which droplet size do you want to use?',
        filter(lambda m: m.memory <= 8192, manager.get_all_sizes()),
        lambda s: f'{s.memory}MiB/{s.disk}GiB@${s.price_monthly}/mo: {s.slug}',
        default='s-1vcpu-1gb',
    ).slug

image = os.environ.get('DROPLET_IMAGE')
if not image:
    image = prompt_select(
        'Which droplet size do you want to use?',
        filter(
            lambda i: (i.slug is not None and region in i.regions and
                       ('ubuntu' in i.slug or 'fedora' in i.slug)),
            manager.get_all_images(),
        ),
        lambda i: f'{i.name}: {i.slug}',
        default='ubuntu-16-04-x64',
    ).slug

# Extract all of the secrets and create runcmds for the initial sync.
cwd = Path(__file__).parent.resolve()
subprocess.run(
    [str(cwd.joinpath('secrets_file_manager.sh')), 'extract'],
    capture_output=True,
)
secrets = cwd.joinpath('secrets')
secrets_runcmds = []
for path in secrets.iterdir():
    with open(path, 'r') as f:
        lines = [
            l.strip().replace('"', r'\"').replace('`', r'\`')
            for l in f.readlines()
        ]
        for line in lines:
            secrets_runcmds.append(
                f'  - echo "{line}" >> /etc/nixos/secrets/{path.name}')

secrets_runcmds = '\n'.join(secrets_runcmds)

user_data = f'''#cloud-config

runcmd:
  - apt install -y git
  - git clone https://git.sr.ht/~sumner/infrastructure /etc/nixos
  - mkdir -p /etc/nixos/secrets
{secrets_runcmds}
  - curl https://raw.githubusercontent.com/elitak/nixos-infect/master/nixos-infect | PROVIDER=digitalocean NIXOS_IMPORT=./host.nix NIX_CHANNEL=nixos-unstable bash 2>&1 | tee /tmp/infect.log
'''

floating_ip_text = '\n' if not floating_ip_to_use else f'''
The following floating IP will be assigned to the machine:
    {floating_ip_to_use.ip}
'''

print()
print('=' * 80)
print('SUMMARY:')
print('=' * 80)
print(f'''
A droplet named "{name}" with initial image of "{image}" and size
"{size}" will be created in the {region} region.

The following SSH keys will be able to access the machine:
    { ', '.join(map(lambda k: k.name, keys_to_use))}
{floating_ip_text}
It will be configured with the following cloud configuration:

{user_data}''')

if not prompt_proceed(
        'Would you like to create a droplet with this configuration?'):
    print('Cancelling!')
    sys.exit(1)

droplet = digitalocean.Droplet(
    backups=False,
    image=image,
    ipv6=True,
    name=name,
    private_networking=True,
    region=region,
    size_slug=size,
    ssh_keys=keys_to_use,
    tags=[],
    token=token,
    user_data=user_data,
)

print('Creating...', end=' ')
droplet.create()

if floating_ip_to_use:
    floating_ip_to_use.assign(droplet.id)

print('DONE')