gnuviech/gvasalt
1
0
Fork 0
Code Issues Pull requests Releases Wiki Activity

Python 3 compatibility for custom states

Browse source
This commit is contained in:
Jan Dittberner 2020-03-04 19:33:23 +01:00
parent bcb92e483d
commit 288acee379
2 changed files with 73 additions and 85 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

109
states/_states/rsa_key.py
View file

@ -9,36 +9,36 @@ import os
def _check_user(user, group):
'''
"""
Checks if the named user and group are present on the minion
'''
err = ''
"""
err = ""
if user:
uid = __salt__['file.user_to_uid'](user)
if uid == '':
err += 'User {0} is not available '.format(user)
uid = __salt__["file.user_to_uid"](user)
if uid == "":
err += "User {0} is not available ".format(user)
if group:
gid = __salt__['file.group_to_gid'](group)
if gid == '':
err += 'Group {0} is not available'.format(group)
gid = __salt__["file.group_to_gid"](group)
if gid == "":
err += "Group {0} is not available".format(group)
return err
def _error(ret, err_msg):
ret['result'] = False
ret['comment'] = err_msg
ret["result"] = False
ret["comment"] = err_msg
return ret
def _calculate_umask(mode):
mode = str(mode).lstrip('0')
mode = str(mode).lstrip("0")
if not mode:
mode = '0'
mode = "0"
modeint = int(mode, 8)
return modeint ^ 0777
return modeint ^ 0o777
def valid_key(name, bits=2048, user=None, group=None, mode='0700'):
def valid_key(name, bits=2048, user=None, group=None, mode="0700"):
"""
Make sure that the given key file exists and contains a valid RSA key.
@ -60,69 +60,62 @@ def valid_key(name, bits=2048, user=None, group=None, mode='0700'):
The permissions set on the file, this defaults to 0600
"""
mode = __salt__['config.manage_mode'](mode)
mode = __salt__["config.manage_mode"](mode)
ret = {
'name': name,
'changes': {},
'result': None,
'comment': ''}
if not os.path.isfile(name) and __opts__['test']:
ret['comment'] = 'would create RSA key in file {0}'.format(name)
ret = {"name": name, "changes": {}, "result": None, "comment": ""}
if not os.path.isfile(name) and __opts__["test"]:
ret["comment"] = "would create RSA key in file {0}".format(name)
return ret
u_check = _check_user(user, group)
if u_check:
return _error(ret, u_check)
if not os.path.isabs(name):
return _error(
ret, 'Specified file {0} is not an absolute path'.format(name))
return _error(ret, "Specified file {0} is not an absolute path".format(name))
if os.path.isdir(name):
return _error(
ret, 'Specified target {0} is a directory'.format(name))
return _error(ret, "Specified target {0} is a directory".format(name))
if os.path.exists(name):
ret, perms = __salt__['file.check_perms'](
name, ret, user, group, mode)
if __opts__['test']:
ret['comment'] = 'File {0} not updated'.format(name)
ret, perms = __salt__["file.check_perms"](name, ret, user, group, mode)
if __opts__["test"]:
ret["comment"] = "File {0} not updated".format(name)
return ret
if not os.path.isfile(name):
rsakey = rsa.generate_private_key(
public_exponent=65537,
key_size=bits,
backend=default_backend())
public_exponent=65537, key_size=bits, backend=default_backend()
)
oldumask = os.umask(_calculate_umask(mode))
with open(name, 'w') as rsafile:
rsafile.write(rsakey.private_bytes(
encoding=serialization.Encoding.PEM,
format=serialization.PrivateFormat.PKCS8,
encryption_algorithm=serialization.NoEncryption()
))
with open(name, "wb") as rsafile:
rsafile.write(
rsakey.private_bytes(
encoding=serialization.Encoding.PEM,
format=serialization.PrivateFormat.PKCS8,
encryption_algorithm=serialization.NoEncryption(),
)
)
os.umask(oldumask)
ret['comment'] = 'created new RSA key and saved PEM file {0}'.format(
name)
ret['changes']['created'] = name
ret['result'] = True
ret["comment"] = "created new RSA key and saved PEM file {0}".format(name)
ret["changes"]["created"] = name
ret["result"] = True
return ret
try:
with open(name, 'r') as rsafile:
with open(name, "rb") as rsafile:
rsakey = serialization.load_pem_private_key(
rsafile.read(),
password=None,
backend=default_backend())
rsafile.read(), password=None, backend=default_backend()
)
except Exception as e:
ret['comment'] = 'error loading RSA key from file {0}: {1}'.format(
name, e)
ret['result'] = False
ret["comment"] = "error loading RSA key from file {0}: {1}".format(name, e)
ret["result"] = False
return ret
if rsakey.key_size < bits:
ret['comment'] = (
'RSA key in {0} is only {1} bits, which is less than the '
'required {2} bits'.format(name, rsakey.key_size, bits))
ret['result'] = False
ret["comment"] = (
"RSA key in {0} is only {1} bits, which is less than the "
"required {2} bits".format(name, rsakey.key_size, bits)
)
ret["result"] = False
else:
ret['comment'] = 'RSA key in file {0} is ok ({1} bits)'.format(
name, rsakey.key_size)
ret['result'] = True
ret["comment"] = "RSA key in file {0} is ok ({1} bits)".format(
name, rsakey.key_size
)
ret["result"] = True
return ret

49
states/_states/x509_certificate.py
View file

@ -1,12 +1,12 @@
# -*- coding: utf8 -*-
'''
"""
Manage X.509 certificate life cycle
===================================
This state is useful for managing X.509 certificates' life cycles.
Copyright (c) 2014, 2016 Jan Dittberner <jan@dittberner.info>
'''
Copyright (c) 2014-2020 Jan Dittberner <jan@dittberner.info>
"""
from cryptography import x509
from cryptography.hazmat.backends import default_backend
@ -15,15 +15,15 @@ import os
def _error(ret, err_msg):
ret['result'] = False
ret['comment'] = err_msg
ret["result"] = False
ret["comment"] = err_msg
return ret
def valid_certificate(
name, mindays=14, keyfile=None,
checkchain=False, trustedcerts=None):
'''
name, mindays=14, keyfile=None, checkchain=False, trustedcerts=None
):
"""
Checks whether the given certificate file is valid.
name
@ -31,33 +31,28 @@ def valid_certificate(
mindays
Mark the certificate as invalid if it is valid for less then this many
days
'''
ret = {
'name': name,
'changes': {},
'result': None,
'comment': ''}
"""
ret = {"name": name, "changes": {}, "result": None, "comment": ""}
if not os.path.isfile(name):
return _error(
ret, 'certificate file {0} does not exist'.format(name))
with open(name) as pemfile:
return _error(ret, "certificate file {0} does not exist".format(name))
with open(name, "rb") as pemfile:
try:
cert = x509.load_pem_x509_certificate(pemfile.read(),
default_backend())
cert = x509.load_pem_x509_certificate(pemfile.read(), default_backend())
except Exception as e:
return _error(
ret, 'error loading certificate {0}: {1}'.format(name, e))
return _error(ret, "error loading certificate {0}: {1}".format(name, e))
notafter = cert.not_valid_after
delta = notafter - datetime.utcnow()
if delta.days < mindays:
return _error(
ret,
'certificate {0} is only valid for {1} more day(s)'.format(
name, delta.days))
"certificate {0} is only valid for {1} more day(s)".format(
name, delta.days
),
)
# TODO: check keyfile match
# TODO: check trust chain
ret['comment'] = (
'certificate {0} is ok and still valid for {1} days'.format(
name, delta.days))
ret['result'] = True
ret["comment"] = "certificate {0} is ok and still valid for {1} days".format(
name, delta.days
)
ret["result"] = True
return ret