tests: Add test for machine attribute updates

This commit is contained in:
Lauri Võsandi 2017-07-07 21:07:25 +00:00
parent 39363a57c7
commit a48331a9cb
6 changed files with 36 additions and 4 deletions

View File

@ -175,7 +175,6 @@ class NormalizeMiddleware(object):
def certidude_app(log_handlers=[]):
from certidude import config
from .revoked import RevocationListResource
from .signed import SignedCertificateDetailResource
from .request import RequestListResource, RequestDetailResource
from .lease import LeaseResource, LeaseDetailResource
@ -191,7 +190,6 @@ def certidude_app(log_handlers=[]):
# Certificate authority API calls
app.add_route("/api/certificate/", CertificateAuthorityResource())
app.add_route("/api/revoked/", RevocationListResource())
app.add_route("/api/signed/{cn}/", SignedCertificateDetailResource())
app.add_route("/api/request/{cn}/", RequestDetailResource())
app.add_route("/api/request/", RequestListResource())
@ -217,6 +215,11 @@ def certidude_app(log_handlers=[]):
# Bootstrap resource
app.add_route("/api/bootstrap/", BootstrapResource())
# Add CRL handler if we have any whitelisted subnets
if config.CRL_SUBNETS:
from .revoked import RevocationListResource
app.add_route("/api/revoked/", RevocationListResource())
# Add SCEP handler if we have any whitelisted subnets
if config.SCEP_SUBNETS:
from .scep import SCEPResource

View File

@ -12,6 +12,7 @@ from oscrypto import keys, asymmetric, symmetric
from oscrypto.errors import SignatureError
class OCSPResource(object):
@whitelist_subnets(config.OCSP_SUBNETS)
def __call__(self, req, resp):
if req.method == "GET":
_, _, _, tail = req.path.split("/", 3)

View File

@ -5,6 +5,7 @@ import json
import logging
from certidude import const, config
from certidude.authority import export_crl, list_revoked
from certidude.firewall import whitelist_subnets
from cryptography import x509
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives.serialization import Encoding
@ -12,6 +13,7 @@ from cryptography.hazmat.primitives.serialization import Encoding
logger = logging.getLogger(__name__)
class RevocationListResource(object):
@whitelist_subnets(config.CRL_SUBNETS)
def on_get(self, req, resp):
# Primarily offer DER encoded CRL as per RFC5280
# This is also what StrongSwan expects

View File

@ -36,6 +36,8 @@ SCEP_SUBNETS = set([ipaddress.ip_network(j) for j in
cp.get("authorization", "scep subnets").split(" ") if j])
OCSP_SUBNETS = set([ipaddress.ip_network(j) for j in
cp.get("authorization", "ocsp subnets").split(" ") if j])
CRL_SUBNETS = set([ipaddress.ip_network(j) for j in
cp.get("authorization", "crl subnets").split(" ") if j])
AUTHORITY_DIR = "/var/lib/certidude"
AUTHORITY_PRIVATE_KEY_PATH = cp.get("authority", "private key path")

View File

@ -62,6 +62,9 @@ scep subnets =
# Online Certificate Status Protocol enabled subnets
ocsp subnets =
# Certificate Revocation lists can be accessed from anywhere by default
crl subnets = 0.0.0.0/0
[logging]
# Disable logging
;backend =

View File

@ -1,4 +1,5 @@
import pwd
from xattr import listxattr, getxattr
from click.testing import CliRunner
from datetime import datetime, timedelta
from time import sleep
@ -154,7 +155,7 @@ def test_cli_setup_authority():
assert not os.environ.get("KRB5_KTNAME"), "Environment contaminated"
# Mock Fedora
for util in "/usr/bin/chcon", "/usr/bin/dnf", "/usr/bin/update-ca-trust":
for util in "/usr/bin/chcon", "/usr/bin/dnf", "/usr/bin/update-ca-trust", "/usr/sbin/dmidecode":
with open(util, "w") as fh:
fh.write("#!/bin/bash\n")
fh.write("exit 0\n")
@ -193,7 +194,7 @@ def test_cli_setup_authority():
# Bootstrap domain controller here,
# Samba startup takes some time
os.system("apt-get install -y samba krb5-user winbind")
os.system("apt-get install -y samba krb5-user winbind bc")
if os.path.exists("/etc/samba/smb.conf"):
os.unlink("/etc/samba/smb.conf")
os.system("samba-tool domain provision --server-role=dc --domain=EXAMPLE --realm=EXAMPLE.LAN --host-name=ca")
@ -308,6 +309,11 @@ def test_cli_setup_authority():
r = requests.get("http://ca.example.lan/api/revoked/")
assert r.status_code == 200, r.text
# Check that SCEP and OCSP are disabled by default
r = requests.get("http://ca.example.lan/api/ocsp/")
assert r.status_code == 404, r.text
r = requests.get("http://ca.example.lan/api/scep/")
assert r.status_code == 404, r.text
# Test command line interface
result = runner.invoke(cli, ['list', '-srv'])
@ -554,6 +560,16 @@ def test_cli_setup_authority():
headers={"Authorization":admintoken})
assert r.status_code == 200, r.text # lease update ok
# Attempt to fetch and execute default.sh script
assert not [j for j in listxattr("/var/lib/certidude/ca.example.lan/signed/test.pem") if j.startswith("user.machine.")]
#os.system("curl http://ca.example.lan/api/signed/test/script | bash")
r = client().simulate_post("/api/signed/test/attr", body="cpu=i5&mem=512M&dist=Ubuntu",
headers={"content-type": "application/x-www-form-urlencoded"})
assert r.status_code == 200, r.text
assert getxattr("/var/lib/certidude/ca.example.lan/signed/test.pem", "user.machine.cpu") == "i5"
assert getxattr("/var/lib/certidude/ca.example.lan/signed/test.pem", "user.machine.mem") == "512M"
assert getxattr("/var/lib/certidude/ca.example.lan/signed/test.pem", "user.machine.dist") == "Ubuntu"
# Test tagging integration in scripting framework
r = client().simulate_get("/api/signed/test/script/")
assert r.status_code == 200, r.text # script render ok
@ -1006,10 +1022,15 @@ def test_cli_setup_authority():
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")
os.system("sed -e 's/crl subnets =.*/crl subnets =/g' -i /etc/certidude/server.conf")
os.system("sed -e 's/address = certificates@example.lan/address =/g' -i /etc/certidude/server.conf")
from certidude.common import pip
pip("asn1crypto certbuilder")
# CRL-s disabled now
r = requests.get("http://ca.example.lan/api/revoked/")
assert r.status_code == 404, r.text
# Update server credential cache
with open("/etc/cron.hourly/certidude") as fh:
cronjob = fh.read()