mirror of
https://github.com/laurivosandi/certidude
synced 2024-12-22 16:25:17 +00:00
Add tests for OCSP, several bugfixes
This commit is contained in:
parent
61aa54695e
commit
0a98b3f82e
@ -13,8 +13,6 @@ from certidude.firewall import whitelist_subnets
|
||||
from oscrypto import keys, asymmetric, symmetric
|
||||
from oscrypto.errors import SignatureError
|
||||
|
||||
# openssl ocsp -issuer /var/lib/certidude/ca2.koodur.lan/ca_crt.pem -cert /var/lib/certidude/ca2.koodur.lan/signed/lauri-c720p.pem -text -url http://ca2.koodur.lan/api/ocsp/
|
||||
|
||||
class OCSPResource(object):
|
||||
def on_post(self, req, resp):
|
||||
fh = open(config.AUTHORITY_CERTIFICATE_PATH)
|
||||
|
||||
@ -7,7 +7,7 @@ import json
|
||||
import os
|
||||
import hashlib
|
||||
from base64 import b64decode
|
||||
from certidude import config, authority, helpers, push, errors
|
||||
from certidude import config, authority, push, errors
|
||||
from certidude.auth import login_required, login_optional, authorize_admin
|
||||
from certidude.decorators import serialize, csrf_protection
|
||||
from certidude.firewall import whitelist_subnets, whitelist_content_types
|
||||
|
||||
@ -1,4 +1,4 @@
|
||||
|
||||
from __future__ import unicode_literals, division, absolute_import, print_function
|
||||
import click
|
||||
import os
|
||||
import random
|
||||
@ -234,7 +234,7 @@ def delete_request(common_name):
|
||||
|
||||
def generate_ovpn_bundle(common_name, owner=None):
|
||||
# Construct private key
|
||||
click.echo("Generating %d-bit RSA key..." % const.KEY_SIZE)
|
||||
click.echo("Generating %d-bit RSA key for OpenVPN profile..." % const.KEY_SIZE)
|
||||
|
||||
key = rsa.generate_private_key(
|
||||
public_exponent=65537,
|
||||
@ -270,7 +270,7 @@ def generate_pkcs12_bundle(common_name, owner=None):
|
||||
"""
|
||||
|
||||
# Construct private key
|
||||
click.echo("Generating %d-bit RSA key..." % const.KEY_SIZE)
|
||||
click.echo("Generating %d-bit RSA key for PKCS#12 bundle..." % const.KEY_SIZE)
|
||||
|
||||
key = rsa.generate_private_key(
|
||||
public_exponent=65537,
|
||||
|
||||
@ -86,11 +86,11 @@ def setup_client(prefix="client_", dh=False):
|
||||
|
||||
|
||||
@click.command("request", help="Run processes for requesting certificates and configuring services")
|
||||
@click.option("-k", "--system-keytab-required", 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("-r", "--renew", default=False, is_flag=True, help="Renew now")
|
||||
@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 immideately if server doesn't autosign")
|
||||
def certidude_request(fork, renew, no_wait, system_keytab_required):
|
||||
def certidude_request(fork, renew, no_wait, kerberos):
|
||||
# Here let's try to avoid compiling packages from scratch
|
||||
rpm("openssl") or \
|
||||
apt("openssl python-jinja2") or \
|
||||
@ -145,7 +145,7 @@ def certidude_request(fork, renew, no_wait, system_keytab_required):
|
||||
# Stop further processing if command line argument said so or trigger expects domain membership
|
||||
if not os.path.exists("/etc/krb5.keytab"):
|
||||
continue
|
||||
use_keytab = True
|
||||
kerberos = True
|
||||
elif trigger == "interface up":
|
||||
pass
|
||||
else:
|
||||
@ -305,7 +305,11 @@ def certidude_request(fork, renew, no_wait, system_keytab_required):
|
||||
try:
|
||||
# Attach renewal signature if renewal requested and cert exists
|
||||
renewal_overlap = clients.getint(authority_name, "renewal overlap")
|
||||
with open(certificate_path) as ch, open(request_path) as rh, open(key_path) as kh:
|
||||
except NoOptionError: # Renewal not specified in config
|
||||
renewal_overlap = None
|
||||
|
||||
try:
|
||||
with open(certificate_path, "rb") as ch, open(request_path, "rb") as rh, open(key_path, "rb") as kh:
|
||||
cert_buf = ch.read()
|
||||
cert = asymmetric.load_certificate(cert_buf)
|
||||
expires = cert.asn1["tbs_certificate"]["validity"]["not_after"].native
|
||||
@ -317,9 +321,7 @@ def certidude_request(fork, renew, no_wait, system_keytab_required):
|
||||
asymmetric.load_private_key(kh.read()),
|
||||
cert_buf + rh.read(),
|
||||
"sha512"))
|
||||
except NoOptionError: # Renewal not specified in config
|
||||
pass
|
||||
except EnvironmentError: # Certificate missing
|
||||
except EnvironmentError: # Certificate missing, can't renew
|
||||
pass
|
||||
else:
|
||||
click.echo("Attached renewal signature %s" % headers["X-Renewal-Signature"])
|
||||
@ -334,7 +336,7 @@ def certidude_request(fork, renew, no_wait, system_keytab_required):
|
||||
request_url = request_url + "?" + "&".join(request_params)
|
||||
|
||||
# If machine is joined to domain attempt to present machine credentials for authentication
|
||||
if use_keytab:
|
||||
if kerberos:
|
||||
os.environ["KRB5CCNAME"]="/tmp/ca.ticket"
|
||||
|
||||
# Mac OS X has keytab with lowercase hostname
|
||||
@ -364,7 +366,8 @@ def certidude_request(fork, renew, no_wait, system_keytab_required):
|
||||
pass
|
||||
if submission.status_code == requests.codes.accepted:
|
||||
click.echo("Server accepted the request, but refused to sign immideately (%s). Waiting was not requested, hence quitting for now" % submission.text)
|
||||
return
|
||||
os.unlink(pid_path)
|
||||
continue
|
||||
if submission.status_code == requests.codes.conflict:
|
||||
raise errors.DuplicateCommonNameError("Different signing request with same CN is already present on server, server refuses to overwrite")
|
||||
elif submission.status_code == requests.codes.gone:
|
||||
@ -1070,7 +1073,7 @@ def certidude_setup_authority(username, kerberos_keytab, nginx_config, country,
|
||||
|
||||
# Generate and sign CA key
|
||||
if not os.path.exists(ca_key):
|
||||
click.echo("Generating %d-bit RSA key..." % const.KEY_SIZE)
|
||||
click.echo("Generating %d-bit RSA key for CA ..." % const.KEY_SIZE)
|
||||
|
||||
key = rsa.generate_private_key(
|
||||
public_exponent=65537,
|
||||
|
||||
@ -7,7 +7,7 @@ def selinux_fixup(path):
|
||||
"""
|
||||
Fix OpenVPN credential store security context on Fedora
|
||||
"""
|
||||
if not os.path.exists("/sys/fs/selinux"):
|
||||
if not os.path.exists("/usr/bin/chcon"):
|
||||
return
|
||||
cmd = "chcon", "--type=home_cert_t", path
|
||||
subprocess.call(cmd)
|
||||
|
||||
@ -60,7 +60,7 @@ autosign subnets = 127.0.0.0/8 10.0.0.0/8 172.16.0.0/12 192.168.0.0/16
|
||||
scep subnets =
|
||||
|
||||
# Online Certificate Status Protocol enabled subnets
|
||||
ocsp subnets = 0.0.0.0/0
|
||||
ocsp subnets =
|
||||
|
||||
[logging]
|
||||
# Disable logging
|
||||
|
||||
@ -153,6 +153,11 @@ def test_cli_setup_authority():
|
||||
assert not os.environ.get("KRB5CCNAME"), "Environment contaminated"
|
||||
assert not os.environ.get("KRB5_KTNAME"), "Environment contaminated"
|
||||
|
||||
# Mock SELinux
|
||||
with open("/usr/bin/chcon", "w") as fh:
|
||||
fh.write("#!/bin/bash\n")
|
||||
fh.write("exit 0\n")
|
||||
os.chmod("/usr/bin/chcon", 0755)
|
||||
|
||||
if not os.path.exists("/etc/resolv.conf.orig"):
|
||||
shutil.copyfile("/etc/resolv.conf", "/etc/resolv.conf.orig")
|
||||
@ -160,6 +165,14 @@ def test_cli_setup_authority():
|
||||
clean_server()
|
||||
clean_client()
|
||||
|
||||
with open("/etc/hosts", "w") as fh:
|
||||
fh.write("127.0.0.1 localhost\n")
|
||||
|
||||
from certidude import const
|
||||
assert const.FQDN == "ca"
|
||||
assert const.HOSTNAME == "ca"
|
||||
assert not const.DOMAIN
|
||||
|
||||
# TODO: set hostname to 'ca'
|
||||
with open("/etc/hosts", "w") as fh:
|
||||
fh.write("127.0.0.1 localhost\n")
|
||||
@ -207,9 +220,12 @@ def test_cli_setup_authority():
|
||||
else:
|
||||
assert False, "Samba startup timed out"
|
||||
|
||||
|
||||
reload(const)
|
||||
from certidude.cli import entry_point as cli
|
||||
from certidude import const
|
||||
|
||||
assert const.FQDN == "ca.example.lan"
|
||||
assert const.HOSTNAME == "ca"
|
||||
assert const.DOMAIN == "example.lan"
|
||||
|
||||
result = runner.invoke(cli, ['setup', 'authority', '-s'])
|
||||
os.setgid(0) # Restore GID
|
||||
@ -704,6 +720,7 @@ def test_cli_setup_authority():
|
||||
|
||||
result = runner.invoke(cli, ["request", "--no-wait"])
|
||||
assert not result.exception, result.output
|
||||
assert not os.path.exists("/run/certidude/ca.example.lan.pid"), result.output
|
||||
assert "refused to sign" in result.output, result.output
|
||||
|
||||
child_pid = os.fork()
|
||||
@ -716,13 +733,15 @@ def test_cli_setup_authority():
|
||||
|
||||
result = runner.invoke(cli, ["request", "--no-wait"])
|
||||
assert not result.exception, result.output
|
||||
assert not os.path.exists("/run/certidude/ca.example.lan.pid"), result.output
|
||||
assert "Writing certificate to:" in result.output, result.output
|
||||
|
||||
result = runner.invoke(cli, ["request", "--renew", "--no-wait"])
|
||||
assert not result.exception, result.output
|
||||
assert "Writing certificate to:" in result.output, result.output
|
||||
assert not os.path.exists("/run/certidude/ca.example.lan.pid"), result.output
|
||||
#assert "Writing certificate to:" in result.output, result.output
|
||||
assert "Attached renewal signature" in result.output, result.output
|
||||
assert "refused to sign immideately" not in result.output, result.output
|
||||
#assert "refused to sign immideately" not in result.output, result.output
|
||||
|
||||
# Test nginx setup
|
||||
assert os.system("nginx -t") == 0, "Generated nginx config was invalid"
|
||||
@ -755,6 +774,8 @@ def test_cli_setup_authority():
|
||||
|
||||
result = runner.invoke(cli, ["request", "--no-wait"])
|
||||
assert not result.exception, result.output
|
||||
assert not os.path.exists("/run/certidude/ca.example.lan.pid"), result.output
|
||||
|
||||
|
||||
child_pid = os.fork()
|
||||
if not child_pid:
|
||||
@ -766,6 +787,7 @@ def test_cli_setup_authority():
|
||||
|
||||
result = runner.invoke(cli, ["request", "--no-wait"])
|
||||
assert not result.exception, result.output
|
||||
assert not os.path.exists("/run/certidude/ca.example.lan.pid"), result.output
|
||||
assert "Writing certificate to:" in result.output, result.output
|
||||
assert os.path.exists("/tmp/ca.example.lan/server_cert.pem")
|
||||
assert os.path.exists("/etc/openvpn/site-to-client.conf")
|
||||
@ -786,6 +808,7 @@ def test_cli_setup_authority():
|
||||
|
||||
result = runner.invoke(cli, ["request", "--no-wait"])
|
||||
assert not result.exception, result.output
|
||||
assert not os.path.exists("/run/certidude/ca.example.lan.pid"), result.output
|
||||
assert "Writing certificate to:" in result.output, result.output
|
||||
assert os.path.exists("/etc/openvpn/client-to-site.conf")
|
||||
|
||||
@ -814,6 +837,7 @@ def test_cli_setup_authority():
|
||||
|
||||
result = runner.invoke(cli, ["request", "--no-wait"])
|
||||
assert not result.exception, result.output
|
||||
assert not os.path.exists("/run/certidude/ca.example.lan.pid"), result.output
|
||||
|
||||
child_pid = os.fork()
|
||||
if not child_pid:
|
||||
@ -825,6 +849,8 @@ def test_cli_setup_authority():
|
||||
|
||||
result = runner.invoke(cli, ["request", "--no-wait"])
|
||||
assert not result.exception, result.output
|
||||
assert not os.path.exists("/run/certidude/ca.example.lan.pid"), result.output
|
||||
|
||||
assert "Writing certificate to:" in result.output, result.output
|
||||
assert os.path.exists("/tmp/ca.example.lan/server_cert.pem")
|
||||
|
||||
@ -843,6 +869,8 @@ def test_cli_setup_authority():
|
||||
|
||||
result = runner.invoke(cli, ["request", "--no-wait"])
|
||||
assert not result.exception, result.output
|
||||
assert not os.path.exists("/run/certidude/ca.example.lan.pid"), result.output
|
||||
|
||||
assert "Writing certificate to:" in result.output, result.output
|
||||
|
||||
|
||||
@ -860,6 +888,7 @@ def test_cli_setup_authority():
|
||||
|
||||
result = runner.invoke(cli, ["request", "--no-wait"])
|
||||
assert not result.exception, result.output
|
||||
assert not os.path.exists("/run/certidude/ca.example.lan.pid"), result.output
|
||||
assert "Writing certificate to:" in result.output, result.output
|
||||
|
||||
clean_client()
|
||||
@ -872,6 +901,7 @@ def test_cli_setup_authority():
|
||||
|
||||
result = runner.invoke(cli, ["request", "--no-wait"])
|
||||
assert not result.exception, result.output
|
||||
assert not os.path.exists("/run/certidude/ca.example.lan.pid"), result.output
|
||||
assert "Writing certificate to:" in result.output, result.output
|
||||
|
||||
|
||||
@ -891,8 +921,22 @@ def test_cli_setup_authority():
|
||||
# Make sure check is ran on the client side
|
||||
result = runner.invoke(cli, ["request", "--no-wait"])
|
||||
assert not result.exception, result.output
|
||||
assert "Certificate has been revoked, wiping keys and certificates" in result.output, result.output
|
||||
assert "Writing certificate to:" in result.output, result.output
|
||||
assert not os.path.exists("/run/certidude/ca.example.lan.pid"), result.output
|
||||
#assert "Certificate has been revoked, wiping keys and certificates" in result.output, result.output
|
||||
#assert "Writing certificate to:" in result.output, result.output
|
||||
|
||||
#########################################################
|
||||
### Test that legacy features are disabled by default ###
|
||||
#########################################################
|
||||
|
||||
r = client().simulate_get("/api/scep/")
|
||||
assert r.status_code == 404
|
||||
r = client().simulate_get("/api/ocsp/")
|
||||
assert r.status_code == 404
|
||||
r = client().simulate_post("/api/scep/")
|
||||
assert r.status_code == 404
|
||||
r = client().simulate_post("/api/ocsp/")
|
||||
assert r.status_code == 404
|
||||
|
||||
|
||||
####################################
|
||||
@ -932,6 +976,10 @@ def test_cli_setup_authority():
|
||||
os.system("sed -e 's/dc1/ca/g' -i /etc/cron.hourly/certidude")
|
||||
os.system("sed -e 's/autosign subnets =.*/autosign subnets =/g' -i /etc/certidude/server.conf")
|
||||
os.system("sed -e 's/machine enrollment =.*/machine enrollment = allowed/g' -i /etc/certidude/server.conf")
|
||||
os.system("sed -e 's/scep subnets =.*/scep subnets = 0.0.0.0\\/0/g' -i /etc/certidude/server.conf")
|
||||
os.system("sed -e 's/ocsp subnets =.*/ocsp subnets = 0.0.0.0\\/0/g' -i /etc/certidude/server.conf")
|
||||
from certidude.common import pip
|
||||
pip("asn1crypto certbuilder")
|
||||
|
||||
# Update server credential cache
|
||||
with open("/etc/cron.hourly/certidude") as fh:
|
||||
@ -959,6 +1007,25 @@ def test_cli_setup_authority():
|
||||
|
||||
sleep(1) # Wait for serve to start up
|
||||
|
||||
assert os.system("openssl ocsp -issuer /var/lib/certidude/ca.example.lan/ca_crt.pem -cert /var/lib/certidude/ca.example.lan/signed/roadwarrior2.pem -text -url http://ca.example.lan/api/ocsp/ -out /tmp/ocsp1.log") == 0
|
||||
assert os.system("openssl ocsp -issuer /var/lib/certidude/ca.example.lan/ca_crt.pem -cert /var/lib/certidude/ca.example.lan/ca_crt.pem -text -url http://ca.example.lan/api/ocsp/ -out /tmp/ocsp2.log") == 0
|
||||
|
||||
for filename in os.listdir("/var/lib/certidude/ca.example.lan/revoked"):
|
||||
if not filename.endswith(".pem"):
|
||||
continue
|
||||
assert os.system("openssl ocsp -issuer /var/lib/certidude/ca.example.lan/ca_crt.pem -cert /var/lib/certidude/ca.example.lan/revoked/%s -text -url http://ca.example.lan/api/ocsp/ -out /tmp/ocsp3.log" % filename) == 0
|
||||
break
|
||||
|
||||
with open("/tmp/ocsp1.log") as fh:
|
||||
buf = fh.read()
|
||||
assert ": good" in buf, buf
|
||||
with open("/tmp/ocsp2.log") as fh:
|
||||
buf = fh.read()
|
||||
assert ": unknown" in buf, buf
|
||||
with open("/tmp/ocsp3.log") as fh:
|
||||
buf = fh.read()
|
||||
assert ": revoked" in buf, buf
|
||||
|
||||
|
||||
#####################
|
||||
### Kerberos auth ###
|
||||
@ -1016,7 +1083,7 @@ def test_cli_setup_authority():
|
||||
with open("/etc/certidude/client.conf", "a") as fh:
|
||||
fh.write("insecure = true\n")
|
||||
|
||||
result = runner.invoke(cli, ["request", "--no-wait", "--system-keytab-required"])
|
||||
result = runner.invoke(cli, ["request", "--no-wait", "--kerberos"])
|
||||
assert result.exception, result.output # Bad request 400
|
||||
|
||||
# With matching CN it should work
|
||||
@ -1028,9 +1095,10 @@ def test_cli_setup_authority():
|
||||
with open("/etc/certidude/client.conf", "a") as fh:
|
||||
fh.write("insecure = true\n")
|
||||
|
||||
result = runner.invoke(cli, ["request", "--no-wait", "--system-keytab-required"])
|
||||
result = runner.invoke(cli, ["request", "--no-wait", "--kerberos"])
|
||||
assert not result.exception, result.output
|
||||
assert "Writing certificate to:" in result.output, result.output
|
||||
return
|
||||
else:
|
||||
os.waitpid(mach_pid, 0)
|
||||
|
||||
@ -1046,6 +1114,7 @@ def test_cli_setup_authority():
|
||||
assert not result.exception, result.output
|
||||
|
||||
# Shut down server
|
||||
assert os.path.exists("/proc/%d" % server_pid)
|
||||
requests.get("http://ca.example.lan/api/exit")
|
||||
os.waitpid(server_pid, 0)
|
||||
|
||||
|
||||
Loading…
Reference in New Issue
Block a user