Fix formatting issues

This commit is contained in:
Lauri Võsandi 2021-06-05 04:50:27 +00:00
parent 22290db3bb
commit d549ab4999
7 changed files with 76 additions and 72 deletions

6
.flake8 Normal file
View File

@ -0,0 +1,6 @@
[flake8]
inline-quotes = "
multiline-quotes = """
indent-size = 4
max-line-length = 160
ignore = Q003 E128 E704 E731

9
.gitlint Normal file
View File

@ -0,0 +1,9 @@
[general]
ignore=body-is-missing,T3
ignore-stdin=true
[title-match-regex]
regex=[A-Z]
[author-valid-email]
regex=[^@]+@pinecrypt.com

11
.pre-commit-config.yaml Normal file
View File

@ -0,0 +1,11 @@
repos:
- repo: https://github.com/PyCQA/flake8
rev: 3.9.2
hooks:
- id: flake8
additional_dependencies: [flake8-typing-imports==1.10.0,flake8-quotes==3.2.0]
- repo: https://github.com/jorisroovers/gitlint
rev: v0.15.1
hooks:
- id: gitlint

View File

@ -1,3 +1,5 @@
# Background # Background
Certidude is the VPN connectivity client for Pinecrypt Gateway Certidude is the VPN connectivity client for Pinecrypt Gateway.
Code snippet for installing the utility and provisioning the
connection is exposed in the Pinecrypt Gateway user interface

View File

@ -2,24 +2,17 @@
import click import click
import hashlib import hashlib
import logging
import ipsecparse import ipsecparse
import json import json
import os import os
import random
import re import re
import signal import signal
import string
import socket
import subprocess import subprocess
import sys
import requests import requests
from asn1crypto import pem, x509 from asn1crypto import pem, x509
from asn1crypto.csr import CertificationRequest from asn1crypto.csr import CertificationRequest
from certbuilder import CertificateBuilder, pem_armor_certificate
from csrbuilder import CSRBuilder, pem_armor_csr from csrbuilder import CSRBuilder, pem_armor_csr
from configparser import ConfigParser, NoOptionError from configparser import ConfigParser, NoOptionError
from datetime import datetime, timedelta
from email.utils import formatdate from email.utils import formatdate
from oscrypto import asymmetric from oscrypto import asymmetric
from pinecrypt.client import const from pinecrypt.client import const
@ -37,8 +30,9 @@ retry = Retry(
status_forcelist=(500, 502, 504), status_forcelist=(500, 502, 504),
) )
adapter = HTTPAdapter(max_retries=retry) adapter = HTTPAdapter(max_retries=retry)
session.mount('http://', adapter) session.mount("http://", adapter)
session.mount('https://', adapter) session.mount("https://", adapter)
def selinux_fixup(path): def selinux_fixup(path):
""" """
@ -90,16 +84,16 @@ def certidude_provision(authority, method):
client_config.set(authority, "authority path", os.path.join(b, "ca_cert.pem")) client_config.set(authority, "authority path", os.path.join(b, "ca_cert.pem"))
if method: if method:
client_config.set(authority, "method", method) client_config.set(authority, "method", method)
with open(const.CLIENT_CONFIG_PATH + ".part", 'w') as fh: with open(const.CLIENT_CONFIG_PATH + ".part", "w") as fh:
client_config.write(fh) client_config.write(fh)
os.rename(const.CLIENT_CONFIG_PATH + ".part", const.CLIENT_CONFIG_PATH) os.rename(const.CLIENT_CONFIG_PATH + ".part", const.CLIENT_CONFIG_PATH)
os.system("certidude enroll") os.system("certidude enroll")
@click.command("enroll", help="Run processes for requesting certificates and configuring services") @click.command("enroll", help="Run processes for requesting certificates and configuring services")
@click.option("-k", "--kerberos", default=False, is_flag=True, help="Offer system keytab for auth") @click.option("-k", "--kerberos", default=False, is_flag=True, help="Offer system keytab for auth")
@click.option("-f", "--fork", default=False, is_flag=True, help="Fork to background") @click.option("-f", "--fork", default=False, is_flag=True, help="Fork to background")
@click.option("-nw", "--no-wait", default=False, is_flag=True, help="Return immediately if server doesn't autosign") @click.option("-nw", "--no-wait", default=False, is_flag=True, help="Return immediately if server doesn't autosign")
def certidude_enroll(fork, no_wait, kerberos): def certidude_enroll(fork, no_wait, kerberos):
try: try:
os.makedirs(const.RUN_DIR) os.makedirs(const.RUN_DIR)
@ -128,10 +122,9 @@ def certidude_enroll(fork, no_wait, kerberos):
else: else:
raise raise
#####################
######################### # Fork if requested #
### Fork if requested ### #####################
#########################
pid_path = os.path.join(const.RUN_DIR, authority_name + ".pid") pid_path = os.path.join(const.RUN_DIR, authority_name + ".pid")
@ -233,10 +226,9 @@ def certidude_enroll(fork, no_wait, kerberos):
if not re.match(const.RE_COMMON_NAME, common_name): if not re.match(const.RE_COMMON_NAME, common_name):
raise ValueError("Supplied common name %s doesn't match the expression %s" % (common_name, const.RE_COMMON_NAME)) raise ValueError("Supplied common name %s doesn't match the expression %s" % (common_name, const.RE_COMMON_NAME))
############################
################################ # Generate keypair and CSR #
### Generate keypair and CSR ### ############################
################################
try: try:
key_path = clients.get(authority_name, "key path") key_path = clients.get(authority_name, "key path")
@ -266,19 +258,18 @@ def certidude_enroll(fork, no_wait, kerberos):
builder = CSRBuilder({"common_name": common_name}, self_public_key) builder = CSRBuilder({"common_name": common_name}, self_public_key)
request = builder.build(private_key) request = builder.build(private_key)
with open(key_partial, 'wb') as f: with open(key_partial, "wb") as f:
f.write(asymmetric.dump_private_key(private_key, None)) f.write(asymmetric.dump_private_key(private_key, None))
with open(request_partial, 'wb') as f: with open(request_partial, "wb") as f:
f.write(pem_armor_csr(request)) f.write(pem_armor_csr(request))
selinux_fixup(key_partial) selinux_fixup(key_partial)
selinux_fixup(request_partial) selinux_fixup(request_partial)
os.rename(key_partial, key_path) os.rename(key_partial, key_path)
os.rename(request_partial, request_path) os.rename(request_partial, request_path)
##########################################
############################################## # Submit CSR and save signed certificate #
### Submit CSR and save signed certificate ### ##########################################
##############################################
try: try:
certificate_path = clients.get(authority_name, "certificate path") certificate_path = clients.get(authority_name, "certificate path")
@ -348,7 +339,8 @@ def certidude_enroll(fork, no_wait, kerberos):
if submission.status_code == requests.codes.ok: if submission.status_code == requests.codes.ok:
pass pass
if submission.status_code == requests.codes.accepted: if submission.status_code == requests.codes.accepted:
click.echo("Server accepted the request, but refused to sign immediately (%s). Waiting was not requested, hence quitting for now" % submission.text) click.echo("Server accepted the request, but refused to sign immediately (%s). "
"Waiting was not requested, hence quitting for now" % submission.text)
os.unlink(pid_path) os.unlink(pid_path)
continue continue
if submission.status_code == requests.codes.conflict: if submission.status_code == requests.codes.conflict:
@ -363,8 +355,8 @@ def certidude_enroll(fork, no_wait, kerberos):
try: try:
header, _, certificate_der_bytes = pem.unarmor(submission.content) header, _, certificate_der_bytes = pem.unarmor(submission.content)
cert = x509.Certificate.load(certificate_der_bytes) x509.Certificate.load(certificate_der_bytes)
except: # TODO: catch correct exceptions except ValueError:
raise ValueError("Failed to parse PEM: %s" % submission.text) raise ValueError("Failed to parse PEM: %s" % submission.text)
os.umask(0o022) os.umask(0o022)
@ -380,10 +372,9 @@ def certidude_enroll(fork, no_wait, kerberos):
else: else:
click.echo("Certificate found at %s and no renewal requested" % certificate_path) click.echo("Certificate found at %s and no renewal requested" % certificate_path)
##############################
################################## # Configure related services #
### Configure related services ### ##############################
##################################
endpoint = authority_name endpoint = authority_name
@ -429,12 +420,7 @@ def certidude_enroll(fork, no_wait, kerberos):
if os.path.exists("/bin/systemctl"): if os.path.exists("/bin/systemctl"):
click.echo("Re-running systemd generators for OpenVPN...") click.echo("Re-running systemd generators for OpenVPN...")
os.system("systemctl daemon-reload") os.system("systemctl daemon-reload")
# if not os.path.exists("/etc/systemd/system/openvpn-reconnect.service"): # TODO: Restore openvpn-reconnect.service here
# with open("/etc/systemd/system/openvpn-reconnect.service.part", "w") as fh:
# fh.write(env.get_template("client/openvpn-reconnect.service").render(context))
# os.rename("/etc/systemd/system/openvpn-reconnect.service.part",
# "/etc/systemd/system/openvpn-reconnect.service")
# click.echo("Created /etc/systemd/system/openvpn-reconnect.service")
os.system("systemctl restart openvpn") os.system("systemctl restart openvpn")
continue continue
@ -458,9 +444,7 @@ def certidude_enroll(fork, no_wait, kerberos):
config["conn", endpoint]["esp"] = "%s!" % bootstrap["strongswan"]["esp"] config["conn", endpoint]["esp"] = "%s!" % bootstrap["strongswan"]["esp"]
config["conn", endpoint]["leftsourceip"] = "%config,%config6" config["conn", endpoint]["leftsourceip"] = "%config,%config6"
config["conn", endpoint]["leftcert"] = certificate_path config["conn", endpoint]["leftcert"] = certificate_path
# leftca="$AUTHORITY_CERTIFICATE_DISTINGUISHED_NAME" # TODO: Assert DN values here?
# rightca="$AUTHORITY_CERTIFICATE_DISTINGUISHED_NAME"
with open(strongswan_secrets_path + ".part", "w") as fh: with open(strongswan_secrets_path + ".part", "w") as fh:
fh.write(": %s %s\n" % ( fh.write(": %s %s\n" % (
@ -537,10 +521,8 @@ def certidude_enroll(fork, no_wait, kerberos):
os.system("nmcli con up %s" % uuid) os.system("nmcli con up %s" % uuid)
continue continue
# IPSec set up with NetworkManager # IPSec set up with NetworkManager
if method == "network-manager/strongswan": if method == "network-manager/strongswan":
client_config = ConfigParser()
nm_config = ConfigParser() nm_config = ConfigParser()
nm_config.add_section("connection") nm_config.add_section("connection")
nm_config.set("connection", "certidude managed", "true") nm_config.set("connection", "certidude managed", "true")
@ -557,7 +539,7 @@ def certidude_enroll(fork, no_wait, kerberos):
nm_config.set("vpn", "userkey", key_path) nm_config.set("vpn", "userkey", key_path)
nm_config.set("vpn", "usercert", certificate_path) nm_config.set("vpn", "usercert", certificate_path)
nm_config.set("vpn", "certificate", authority_path) nm_config.set("vpn", "certificate", authority_path)
nm_config.set("vpn", "ike", bootstrap["strongswan"]["ike"]) nm_config.set("vpn", "ike", bootstrap["strongswan"]["ike"]) # TODO: Check if the ! syntax is used
nm_config.set("vpn", "esp", bootstrap["strongswan"]["esp"]) nm_config.set("vpn", "esp", bootstrap["strongswan"]["esp"])
nm_config.set("vpn", "proposal", "yes") nm_config.set("vpn", "proposal", "yes")
@ -576,7 +558,7 @@ def certidude_enroll(fork, no_wait, kerberos):
os.system("nmcli con up %s" % uuid) os.system("nmcli con up %s" % uuid)
continue continue
click.echo("Unknown service: %s" % service_config.get(endpoint, "service")) click.echo("Unknown provisioning method: %s" % method)
os.unlink(pid_path) os.unlink(pid_path)

View File

@ -4,16 +4,12 @@ import socket
RUN_DIR = "/run/certidude" RUN_DIR = "/run/certidude"
CONFIG_DIR = "/etc/certidude" CONFIG_DIR = "/etc/certidude"
CLIENT_CONFIG_PATH = os.path.join(CONFIG_DIR, "client.conf") CLIENT_CONFIG_PATH = os.path.join(CONFIG_DIR, "client.conf")
SERVICES_CONFIG_PATH = os.path.join(CONFIG_DIR, "services.conf")
RE_FQDN = "^(([a-z0-9]|[a-z0-9][a-z0-9\-_]*[a-z0-9])\.)+([a-z0-9]|[a-z0-9][a-z0-9\-_]*[a-z0-9])?$" RE_FQDN = r"^(([a-z0-9]|[a-z0-9][a-z0-9\-_]*[a-z0-9])\.)+([a-z0-9]|[a-z0-9][a-z0-9\-_]*[a-z0-9])?$"
RE_HOSTNAME = "^[a-z0-9]([a-z0-9\-_]{0,61}[a-z0-9])?$" RE_HOSTNAME = r"^[a-z0-9]([a-z0-9\-_]{0,61}[a-z0-9])?$"
RE_COMMON_NAME = "^[A-Za-z0-9\-\.\_@]+$" RE_COMMON_NAME = r"^[A-Za-z0-9\-\.\_@]+$"
try: FQDN = socket.getfqdn()
FQDN = socket.getaddrinfo(socket.gethostname(), 0, socket.AF_INET, 0, 0, socket.AI_CANONNAME)[0][3]
except socket.gaierror:
FQDN = socket.gethostname()
try: try:
HOSTNAME, DOMAIN = FQDN.split(".", 1) HOSTNAME, DOMAIN = FQDN.split(".", 1)

View File

@ -1,6 +1,5 @@
#!/usr/bin/env python3 #!/usr/bin/env python3
# coding: utf-8 # coding: utf-8
import os
from setuptools import setup from setuptools import setup
setup( setup(
@ -40,4 +39,3 @@ setup(
"Programming Language :: Python :: 3 :: Only", "Programming Language :: Python :: 3 :: Only",
], ],
) )