scripts: Check certificate validity and trustworthiness.
pisign utility embeds certificate into Zip files and validates them before checking signature. It also looks into a trust_dir database to see if certificates are trusted or not. Check README for examples.
This commit is contained in:
@@ -21,11 +21,17 @@ cp test.zip test2.zip
|
||||
|
||||
# Sign ZIP files:
|
||||
|
||||
./pisign.py sign certs/enc_key.pem test.zip test2.zip
|
||||
./pisign.py sign certs/enc_key.pem test.zip test2.zip abc.txt
|
||||
Password: <Enter password>
|
||||
|
||||
# Make a directory for trusted certificates:
|
||||
|
||||
mkdir certs_trusted/
|
||||
cp certs/cert.pem certs_trusted/
|
||||
|
||||
# Verify ZIP files:
|
||||
|
||||
./pisign.py verify certs/cert.pem test.zip test2.zip
|
||||
./pisign.py verify certs_trusted/ test.zip test2.zip abc.txt
|
||||
|
||||
# Change *.zip contents and try to verify again.
|
||||
# Clear cert_trusted/ directory and try again.
|
||||
|
||||
@@ -15,21 +15,33 @@ import sys
|
||||
import tempfile
|
||||
import zipfile
|
||||
|
||||
# ZipFile extensions
|
||||
ZIP_FILES = ('.zip', '.pisi')
|
||||
|
||||
# Signature headers & extensions
|
||||
HEADER = 'pisi-signed'
|
||||
EXT_SIGN = 'sig'
|
||||
EXT_CERT = 'crt'
|
||||
|
||||
# Signature validity
|
||||
SIGN_OK, SIGN_NO, SIGN_SELF, SIGN_UNTRUSTED, SIGN_CORRUPTED = range(5)
|
||||
|
||||
# Certificate validity
|
||||
VALID = 1
|
||||
SELF_SIGNED = 0
|
||||
INVALID = -1
|
||||
CERT_OK, CERT_SELF, CERT_CORRUPTED = range(3)
|
||||
|
||||
# Certificate trustworthiness
|
||||
CERT_TRUSTED, CERT_UNTRUSTED = range(2)
|
||||
|
||||
def sign_data(data, key_file, password_fd):
|
||||
"""
|
||||
Signs data with given certificate.
|
||||
Signs data with given key.
|
||||
|
||||
Arguments:
|
||||
data: Data to be signed
|
||||
key_file: Private key
|
||||
password_fd: File that contains passphrase
|
||||
Returns:
|
||||
Signed data
|
||||
Signed data (binary)
|
||||
"""
|
||||
# Go to begining of password file
|
||||
password_fd.seek(0)
|
||||
@@ -48,10 +60,7 @@ def sign_data(data, key_file, password_fd):
|
||||
# Get signed data
|
||||
signed_binary = pipe.stdout.read()
|
||||
|
||||
# Convert to Base 64
|
||||
signed_ascii = base64.b64encode(signed_binary)
|
||||
|
||||
return signed_ascii
|
||||
return signed_binary
|
||||
|
||||
def get_public_key(cert_file):
|
||||
"""
|
||||
@@ -63,7 +72,7 @@ def get_public_key(cert_file):
|
||||
Public key
|
||||
"""
|
||||
# Use OpenSSL to extract public key
|
||||
command = "openssl x509 -inform pem -in %s -pubkey -noout"
|
||||
command = 'openssl x509 -inform pem -in %s -pubkey -noout'
|
||||
command = command % cert_file
|
||||
command = shlex.split(command)
|
||||
|
||||
@@ -73,6 +82,57 @@ def get_public_key(cert_file):
|
||||
|
||||
return key_ascii
|
||||
|
||||
def get_hash(cert_file):
|
||||
"""
|
||||
Extracts hash from certificate.
|
||||
|
||||
Arguments:
|
||||
cert_file: Certificate
|
||||
Returns:
|
||||
Hash
|
||||
"""
|
||||
# Use OpenSSL to extract hash
|
||||
command = 'openssl x509 -noout -in %s -hash'
|
||||
command = command % cert_file
|
||||
command = shlex.split(command)
|
||||
|
||||
pipe = subprocess.Popen(command, stdout=subprocess.PIPE,
|
||||
stderr=subprocess.PIPE)
|
||||
hash = pipe.stdout.read()
|
||||
hash = hash.strip()
|
||||
|
||||
return hash
|
||||
|
||||
def check_trust(cert_file, trust_dir):
|
||||
"""
|
||||
Checks if certificate is trusted or not.
|
||||
|
||||
Arguments:
|
||||
cert_file: Certificate
|
||||
trust_dir: Path to trust database.
|
||||
Returns:
|
||||
CERT_TRUSTED or CERT_UNTRUSTED
|
||||
"""
|
||||
|
||||
cert_hash = get_hash(cert_file)
|
||||
|
||||
for filename in os.listdir(trust_dir):
|
||||
cert_path = os.path.join(trust_dir, filename)
|
||||
if os.path.exists(cert_path):
|
||||
if cert_hash == get_hash(cert_path):
|
||||
return CERT_TRUSTED
|
||||
return CERT_UNTRUSTED
|
||||
|
||||
"""
|
||||
# Code to be used in production:
|
||||
cert_hash = get_hash(cert_file)
|
||||
cert_path = os.path.join(trust_dir, cert_hash)
|
||||
if os.path.exists(cert_path):
|
||||
if cert_hash == get_hash(cert_path):
|
||||
return CERT_TRUSTED
|
||||
return CERT_UNTRUSTED
|
||||
"""
|
||||
|
||||
def verify_certificate(cert_file):
|
||||
"""
|
||||
Verifies a certificate.
|
||||
@@ -80,7 +140,7 @@ def verify_certificate(cert_file):
|
||||
Arguments:
|
||||
cert_file: Certificate
|
||||
Returns:
|
||||
VALID, SELF_SIGNED or INVALID
|
||||
CERT_OK, CERT_SELF or CERT_CORRUPTED
|
||||
"""
|
||||
# Use OpenSSL to verify certificate
|
||||
command = '/usr/bin/openssl verify %s'
|
||||
@@ -91,36 +151,52 @@ def verify_certificate(cert_file):
|
||||
stderr=subprocess.PIPE)
|
||||
lines = pipe.stdout.read().split('\n')
|
||||
if len(lines) < 2:
|
||||
return INVALID
|
||||
return CERT_CORRUPTED
|
||||
elif lines[1].startswith("error"):
|
||||
code = lines[1].split()[1]
|
||||
if code == '18':
|
||||
return SELF_SIGNED
|
||||
return CERT_SELF
|
||||
else:
|
||||
return INVALID
|
||||
return CERT_CORRUPTED
|
||||
else:
|
||||
return VALID
|
||||
return CERT_OK
|
||||
|
||||
def verify_data(data, signature, cert_file):
|
||||
def verify_file(data_file, cert_file=None, signature_file=None, trust_dir=None):
|
||||
"""
|
||||
Verifies signature of data signed with given certificate.
|
||||
Verifies signature of file signed with given certificate.
|
||||
|
||||
If signature_file is not defined data_file + ".sig" will
|
||||
be used.
|
||||
|
||||
If cert_file is not defined data_file + ".crt" will
|
||||
be used.
|
||||
|
||||
Arguments:
|
||||
data: Original data
|
||||
signature_file: Signed data
|
||||
cert_file: Certificate
|
||||
data_file: Original data file
|
||||
cert_file: Certificate (or None)
|
||||
signature_file: Signature file (or None)
|
||||
trust_dir: Path to trust database.
|
||||
Returns:
|
||||
True if valid, False if invalid
|
||||
SIGN_OK, SIGN_NO, SIGN_SELF or SIGN_CORRUPTED
|
||||
"""
|
||||
# Keep signature in a temporary file
|
||||
signature_file = tempfile.NamedTemporaryFile()
|
||||
signature_file.write(signature)
|
||||
signature_file.flush()
|
||||
# Sanitize before appending signature extension
|
||||
data_file = os.path.realpath(data_file)
|
||||
|
||||
# Keep data in a temporary file
|
||||
data_file = tempfile.NamedTemporaryFile()
|
||||
data_file.write(data)
|
||||
data_file.flush()
|
||||
if not signature_file:
|
||||
signature_file = data_file + '.' + EXT_SIGN
|
||||
if not cert_file:
|
||||
cert_file = data_file + '.' + EXT_CERT
|
||||
if not os.path.exists(signature_file) or not os.path.exists(cert_file):
|
||||
return SIGN_NO
|
||||
|
||||
# Verify certificate
|
||||
cert_validity = verify_certificate(cert_file)
|
||||
if cert_validity == CERT_CORRUPTED:
|
||||
return SIGN_CORRUPTED
|
||||
|
||||
# Check trustworthiness of certificate
|
||||
if trust_dir != None and check_trust(cert_file, trust_dir) == CERT_UNTRUSTED:
|
||||
return SIGN_UNTRUSTED
|
||||
|
||||
# Keep public key in a temporary file
|
||||
pub_file = tempfile.NamedTemporaryFile()
|
||||
@@ -129,7 +205,7 @@ def verify_data(data, signature, cert_file):
|
||||
|
||||
# Use OpenSSL to verify signature
|
||||
command = '/usr/bin/openssl dgst -sha1 -verify %s -signature %s %s'
|
||||
command = command % (pub_file.name, signature_file.name, data_file.name)
|
||||
command = command % (pub_file.name, signature_file, data_file)
|
||||
command = shlex.split(command)
|
||||
|
||||
pipe = subprocess.Popen(command, stdout=subprocess.PIPE,
|
||||
@@ -137,11 +213,64 @@ def verify_data(data, signature, cert_file):
|
||||
result = pipe.wait()
|
||||
|
||||
# Destroy temporary files
|
||||
signature_file.close()
|
||||
data_file.close()
|
||||
pub_file.close()
|
||||
|
||||
return result == 0
|
||||
if result == 0:
|
||||
if cert_validity == CERT_OK:
|
||||
return SIGN_OK
|
||||
else:
|
||||
return SIGN_SELF
|
||||
else:
|
||||
return SIGN_CORRUPTED
|
||||
|
||||
def verify_data(data, signature_data, trust_dir):
|
||||
"""
|
||||
Verifies signature of data signed with given certificate.
|
||||
|
||||
Arguments:
|
||||
data: Original data
|
||||
signature_data: Signature data from ZipFile
|
||||
trust_dir: Path to trust database.
|
||||
Returns:
|
||||
SIGN_OK, SIGN_NO, SIGN_SELF or SIGN_CORRUPTED
|
||||
"""
|
||||
# Check header
|
||||
if not len(signature_data) or not signature_data.startswith(HEADER):
|
||||
return SIGN_NO
|
||||
else:
|
||||
try:
|
||||
header, cert_ascii, signature_ascii = signature_data.split(':')
|
||||
except ValueError:
|
||||
return SIGN_CORRUPTED
|
||||
if header != HEADER:
|
||||
return SIGN_CORRUPTED
|
||||
signature_binary = base64.b64decode(signature_ascii)
|
||||
cert_data = base64.b64decode(cert_ascii)
|
||||
|
||||
# Keep certificate in a temporary file
|
||||
cert_file = tempfile.NamedTemporaryFile()
|
||||
cert_file.write(cert_data)
|
||||
cert_file.flush()
|
||||
|
||||
# Keep signature in a temporary file
|
||||
signature_file = tempfile.NamedTemporaryFile()
|
||||
signature_file.write(signature_binary)
|
||||
signature_file.flush()
|
||||
|
||||
# Keep data in a temporary file
|
||||
data_file = tempfile.NamedTemporaryFile()
|
||||
data_file.write(data)
|
||||
data_file.flush()
|
||||
|
||||
# Verify
|
||||
result = verify_file(data_file.name, cert_file.name, signature_file.name, trust_dir)
|
||||
|
||||
# Destroy temporary files
|
||||
cert_file.close()
|
||||
signature_file.close()
|
||||
data_file.close()
|
||||
|
||||
return result
|
||||
|
||||
def get_zip_hashes(zip_obj):
|
||||
"""
|
||||
@@ -166,15 +295,15 @@ def get_zip_hashes(zip_obj):
|
||||
|
||||
return "\n".join(hashes)
|
||||
|
||||
def verify_zipfile(filename, cert_file):
|
||||
def verify_zipfile(filename, trust_dir=None):
|
||||
"""
|
||||
Verifies integrity of a ZIP file.
|
||||
|
||||
Arguments:
|
||||
filename: ZIP filename
|
||||
cert_file: Certificate
|
||||
trust_dir: Path to trust database.
|
||||
Returns:
|
||||
True if valid, False if invalid
|
||||
SIGNED, UNSIGNED, SELF_SIGNED or CORRUPTED
|
||||
"""
|
||||
|
||||
try:
|
||||
@@ -186,31 +315,57 @@ def verify_zipfile(filename, cert_file):
|
||||
hashes = get_zip_hashes(zip_obj)
|
||||
|
||||
# Read signed hash data from ZIP comment
|
||||
signature = base64.b64decode(zip_obj.comment)
|
||||
signature_data = zip_obj.comment
|
||||
|
||||
# Close ZIP file
|
||||
zip_obj.close()
|
||||
|
||||
# Verify signed data
|
||||
return verify_data(hashes, signature, cert_file)
|
||||
return verify_data(hashes, signature_data, trust_dir)
|
||||
|
||||
def sign_zipfile(filename, certificate, password_fd):
|
||||
def sign_file(filename, key_file, cert_file, password_fd):
|
||||
"""
|
||||
Signs ZIP file with given certificate.
|
||||
Signs file with given key.
|
||||
|
||||
Arguments:
|
||||
filename: File name to be signed
|
||||
certificate: Private certificate
|
||||
key_file: Private key
|
||||
cert_file: Certificate
|
||||
password_fd: File that contains passphrase
|
||||
"""
|
||||
data = file(filename).read()
|
||||
signed_binary = sign_data(data, key_file, password_fd)
|
||||
cert_data = file(cert_file).read()
|
||||
|
||||
# Save certificate
|
||||
file('%s.%s' % (filename, EXT_CERT), 'w').write(cert_data)
|
||||
|
||||
# Save signed data
|
||||
file('%s.%s' % (filename, EXT_SIGN), 'w').write(signed_binary)
|
||||
|
||||
def sign_zipfile(filename, key_file, cert_file, password_fd):
|
||||
"""
|
||||
Signs ZIP file with given key.
|
||||
|
||||
Arguments:
|
||||
filename: File name to be signed
|
||||
key_file: Private key
|
||||
cert_file: Certificate
|
||||
password_fd: File that contains passphrase
|
||||
"""
|
||||
zip_obj = zipfile.ZipFile(filename, 'a')
|
||||
|
||||
# Get ZIP hashes and sign them
|
||||
hashes = get_zip_hashes(zip_obj)
|
||||
hashes_signed = sign_data(hashes, certificate, password_fd)
|
||||
signed_binary = sign_data(hashes, key_file, password_fd)
|
||||
signed_ascii = base64.b64encode(signed_binary)
|
||||
|
||||
# Encode certificate
|
||||
cert_data = file(cert_file).read()
|
||||
cert_ascii = base64.b64encode(cert_data)
|
||||
|
||||
# Add signed data as ZIP comment
|
||||
zip_obj.comment = hashes_signed
|
||||
zip_obj.comment = '%s:%s:%s' % (HEADER, cert_ascii, signed_ascii)
|
||||
|
||||
# Mark file as modified and save it
|
||||
zip_obj._didModify = True
|
||||
@@ -222,8 +377,8 @@ def print_usage():
|
||||
"""
|
||||
|
||||
print "Usage:"
|
||||
print " %s sign <path/to/private_key> <file.zip ...>" % sys.argv[0]
|
||||
print " %s verify <path/to/certificate> <file.zip ...>" % sys.argv[0]
|
||||
print " %s sign <priv_key> <cert> <file1 ...>" % sys.argv[0]
|
||||
print " %s verify <trust_dir> <file1 ...>" % sys.argv[0]
|
||||
sys.exit(1)
|
||||
|
||||
def main():
|
||||
@@ -239,18 +394,22 @@ def main():
|
||||
if operation == 'sign':
|
||||
try:
|
||||
key_file = sys.argv[2]
|
||||
cert_file = sys.argv[3]
|
||||
except IndexError:
|
||||
print_usage()
|
||||
|
||||
if len(sys.argv[3:]):
|
||||
if len(sys.argv[4:]):
|
||||
# Keep password in a temporary file
|
||||
password = getpass.getpass()
|
||||
password_fd = os.tmpfile()
|
||||
password_fd.write(password)
|
||||
password_fd.flush()
|
||||
|
||||
for filename in sys.argv[3:]:
|
||||
sign_zipfile(filename, key_file, password_fd)
|
||||
for filename in sys.argv[4:]:
|
||||
if filename.endswith(ZIP_FILES):
|
||||
sign_zipfile(filename, key_file, cert_file, password_fd)
|
||||
else:
|
||||
sign_file(filename, key_file, cert_file, password_fd)
|
||||
print "Signed %s with %s" % (filename, key_file)
|
||||
|
||||
# Destroy temporary file
|
||||
@@ -260,21 +419,24 @@ def main():
|
||||
|
||||
elif operation == 'verify':
|
||||
try:
|
||||
cert_file = sys.argv[2]
|
||||
trust_dir = sys.argv[2]
|
||||
except IndexError:
|
||||
print_usage()
|
||||
|
||||
cert_validity = verify_certificate(cert_file)
|
||||
if cert_validity == SELF_SIGNED:
|
||||
print "WARNING: Certificate is self-signed."
|
||||
elif cert_validity == INVALID:
|
||||
print "ERROR: Certificate is invalid"
|
||||
sys.exit(1)
|
||||
|
||||
if len(sys.argv[3:]):
|
||||
for filename in sys.argv[3:]:
|
||||
if verify_zipfile(filename, cert_file):
|
||||
print "%s is valid." % filename
|
||||
if filename.endswith(ZIP_FILES):
|
||||
result = verify_zipfile(filename, trust_dir)
|
||||
else:
|
||||
result = verify_file(filename, trust_dir)
|
||||
if result == SIGN_OK:
|
||||
print "%s is signed by a trusted source." % filename
|
||||
elif result == SIGN_NO:
|
||||
print "%s is unsigned." % filename
|
||||
elif result == SIGN_SELF:
|
||||
print "%s is self-signed by a trusted source." % filename
|
||||
elif result == SIGN_UNTRUSTED:
|
||||
print "%s is signed by an untrusted source." % filename
|
||||
else:
|
||||
print "%s is corrupted." % filename
|
||||
else:
|
||||
|
||||
Reference in New Issue
Block a user