ssl-update/update_cert.py

#!/usr/bin/env python3
"""Automation for cert renewal.

assumptions:
  * router is danknasty
    * generate key in /root/.ssh/id_autofirewall on server
    * firewall has config from `authorized_keys` file
    * firewall sshd config contains: `AcceptEnv ssl_service state`
    * firewall has `ssl-update.sh` copied to /usr/local/bin and chmod +x
"""

import getpass
import os
import pathlib
import pexpect
import pwd
import subprocess
import sys
import time

services = ['git', 'plex', 'jellyfin', 'photoprism']

users = {
    'jellyfin': 'jellyfin',
    'git': 'gitea',
    'plex': 'plex',
    'photoprism': 'photoprism',
}

service_systemd = {
    'jellyfin': 'jellyfin',
    'git': 'gitea',
    'plex': 'plexmediaserver',
    'photoprism': 'photoprism',
}

cert_files = ['privkey1.pem', 'fullchain1.pem', 'chain1.pem', 'cert1.pem']


def rmdir(directory):
  directory = pathlib.Path(directory)
  for item in directory.iterdir():
    if item.is_dir():
      rmdir(item)
    else:
      item.unlink()
  directory.rmdir()


def jellyfin():
  cmd = ['/usr/bin/su', '-l', 'luke', '/usr/bin/pass', 'show', 'ssl/jellyfin']
  p = subprocess.run(cmd, capture_output=True)
  export_pw = p.stdout.decode('UTF-8').strip()
  if not export_pw:
    sys.exit('Couldnt get ssl export password for jellyfin')
  cmd = ['/usr/bin/openssl', 'pkcs12', '-export', '-out',
         '/etc/letsencrypt/live/jellyfin.drheck.dev/jellyfin.pfx',
         '-inkey', '/etc/letsencrypt/live/jellyfin.drheck.dev/privkey.pem',
         '-in', '/etc/letsencrypt/live/jellyfin.drheck.dev/cert.pem',
         '-certfile', '/etc/letsencrypt/live/jellyfin.drheck.dev/chain.pem']
  p = pexpect.spawnu(' '.join(cmd))
  res = p.expect(['Enter Export Password:', pexpect.EOF, pexpect.TIMEOUT])
  if res > 0:
    sys.exit('Failed to run openssl to generate '
             f'pkcs12 keys for jellyfin: {p.before}')
  p.sendline(export_pw)
  res = p.expect(['Verifying - Enter Export Password:',
                  pexpect.EOF, pexpect.TIMEOUT])
  if res > 0:
    sys.exit('Failed to run openssl to generate '
             f'pkcs12 keys for jellyfin: {p.before}')
  p.sendline(export_pw)
  res = p.expect([pexpect.EOF, pexpect.TIMEOUT])
  if res > 0:
    sys.exit(f'Failed to run openssl to generate '
              'pkcs12 keys for jellyfin: {p.before}')


def main(args):

  # Get SSH decryption password
  cmd = ['/usr/bin/su', '-l', 'luke', '/usr/bin/pass', 'show', 'ssh/autofirewall']
  p = subprocess.run(cmd, capture_output=True)
  decrypt_pp = p.stdout.decode('UTF-8').strip()
  if not decrypt_pp:
    sys.exit('Couldnt get decryption passpharase')

  # Start Certbot, get data to send to service
  challenge_path = pathlib.Path('/usr/share/nginx/html/'
                                '.well-known/acme-challenge/')
  if challenge_path.is_dir():
    rmdir(challenge_path)
  challenge_path.mkdir()
  challenge_path.chmod(0o755)
  if len(args) != 1:
    sys.exit(f'Give a service to renew: {", ".join(services)} ')
  service = args[0]
  if service not in services:
    sys.exit(f'Give a service to renew: {", ".join(services)} ')

  fqdn = f'{service}.drheck.dev'
  cmd = ['/usr/bin/certbot', 'certonly', '--manual', '-d', fqdn]
  cb = pexpect.spawnu(' '.join(cmd))
  res = cb.expect(['Create a file containing just this data:\r\n\r\n([^\r]+)\r',
                  pexpect.TIMEOUT, pexpect.EOF], timeout=20)
  if res > 0:
    sys.exit('Timed out')
  data = cb.match.group(1)
  long_match = ('And make it available on your web server at this URL:'
                '\r\n\r\nhttp://%s/.well-known/acme-challenge/([^\r]+)\r')
  res = cb.expect([long_match % (fqdn,), pexpect.TIMEOUT, pexpect.EOF])
  if res > 0:
    sys.exit('Timed out')
  filename = cb.match.group(1)
  res = cb.expect(['Press Enter to Continue', pexpect.EOF], timeout=0)

# put data in acme-challenge file
  data_file = challenge_path / pathlib.Path(filename)
  try:
    with open(data_file, 'w') as f:
      f.write(data)
  except:
    sys.exit(f'Failed to write {data_file}')
  data_file.chmod(0o644)

# put symlink in nginx enabled sites
  symlink_name = pathlib.Path(f'{service}-le')
  nx_conf = pathlib.Path('/etc/nginx')
  avail_path = nx_conf / pathlib.Path('sites-available')
  enabled_path = nx_conf / pathlib.Path('sites-enabled')

  service_available_file = avail_path / symlink_name
  service_enabled_symlink = enabled_path / symlink_name
  if not service_enabled_symlink.is_symlink():
    service_enabled_symlink.symlink_to(service_available_file)

# open port 80 to ${service}.drheck.dev
  os.environ['state'] = 'HTTP_UP'
  os.environ['ssl_service'] = service
  cmd = ['/usr/bin/ssh', '-i', '/root/.ssh/id_autofirewall', '-o',
         'SendEnv=state', '-o', 'SendEnv=ssl_service', '-l', 'luke131',
         'danknasty', 'doas', '-n', '/usr/local/bin/ssl-update.sh']
  print(f'cmd: {cmd}')
  p = pexpect.spawnu(' '.join(cmd))
  p.logfile = sys.stderr
  res = p.expect(["Enter passphrase for key '/root/.ssh/id_autofirewall':",
                  pexpect.TIMEOUT, pexpect.EOF])
  if res > 0:
   sys.exit('Couldnt send decryption key to ssh.')
  p.sendline(decrypt_pp)
  res = p.expect(['success', pexpect.TIMEOUT, pexpect.EOF])
  if res > 0:
    sys.exit(f'Failed. error: {p.before}')

  print(f'Turned up HTTP for {service}')

# restart nginx
  cmd = ['/usr/bin/systemctl', 'restart', 'nginx']
  p = subprocess.run(cmd, capture_output=True)
  time.sleep(5)

# get nginx status
  cmd = ['/usr/bin/systemctl', 'status', 'nginx']
  p = subprocess.run(cmd, capture_output=True)
  stdout = p.stdout.decode('UTF-8').split('\n')
  success = False
  for line in stdout:
    if 'Active: active (running)' in line:
      success = True
  if not success:
    sys.exit('nginx did not restart properly')
  print('nginx restarted properly')

# continue with certbot
  cb.sendline()
  res = cb.expect([pexpect.EOF])
  print(cb.before)

# close port 80 to ${service}.drheck.dev
  os.environ['state'] = 'HTTP_DOWN'
  os.environ['ssl_service'] = service
  cmd = ['/usr/bin/ssh', '-i', '/root/.ssh/id_autofirewall', '-o',
         'SendEnv=state', '-o', 'SendEnv=ssl_service', '-l', 'luke131',
         'danknasty', 'doas', '-n', '/usr/local/bin/ssl-update.sh']
  print(f'cmd: {cmd}')
  p = pexpect.spawnu(' '.join(cmd))
  p.logfile = sys.stderr
  res = p.expect(["Enter passphrase for key '/root/.ssh/id_autofirewall':",
                  pexpect.TIMEOUT, pexpect.EOF])
  if res > 0:
   sys.exit('Couldnt send decryption key to ssh.')
  p.sendline(decrypt_pp)
  res = p.expect(['success', pexpect.TIMEOUT, pexpect.EOF])
  if res > 0:
    sys.exit(f'Failed. error: {p.before}')

  print(f'Turned down HTTP for {service}')

# remove symlink in nginx enabled sites
  service_enabled_symlink.unlink()

# restart nginx
  cmd = ['/usr/bin/systemctl', 'restart', 'nginx']
  p = subprocess.run(cmd, capture_output=True)
  time.sleep(5)

# get nginx status
  cmd = ['/usr/bin/systemctl', 'status', 'nginx']
  p = subprocess.run(cmd, capture_output=True)
  stdout = p.stdout.decode('UTF-8').split('\n')
  success = False
  for line in stdout:
    if 'Active: active (running)' in line:
      success = True
  if not success:
    sys.exit('nginx did not restart properly')
  print('nginx restarted properly')

# chmod
  key_path = pathlib.Path('/etc/letsencrypt')
  live = key_path / pathlib.Path(f'live/{service}.drheck.dev')
  archive = key_path / pathlib.Path(f'archive/{service}.drheck.dev')

  user = users[service]
  uid = pwd.getpwnam(user).pw_uid
  gid = pwd.getpwnam(user).pw_gid

  os.chown(live, uid, gid)
  os.chown(archive, uid, gid)
  for cert_file in cert_files:
    os.chown(archive / pathlib.Path(cert_file), uid, gid)

# service specific work
  if service in globals():
    eval(f'{service}()')
  print(f'{service} specific work complete.')

 # restart $service
  systemd = service_systemd[service]
  cmd = ['/usr/bin/systemctl', 'restart', systemd]
  p = subprocess.run(cmd)
  time.sleep(10)

# get $service status
  cmd = ['/usr/bin/systemctl', 'status', systemd]
  p = subprocess.run(cmd, capture_output=True)
  stdout = p.stdout.decode('UTF-8').split('\n')
  success = False
  for line in stdout:
    if 'Active: active (running)' in line:
      success = True
  if not success:
    sys.exit(f'{service} did not restart properly')
  print(f'{service} restarted properly')


if __name__ == '__main__':
  main(sys.argv[1:])