validators.py at master ⋅ cyborgbackup/cyborgbackup

# Python
import base64
import re

# Django
from django.utils.translation import ugettext_lazy as _
from django.core.exceptions import ValidationError

# REST framework
from rest_framework.serializers import ValidationError as RestValidationError
from rest_framework.exceptions import ParseError

# CyBorgBackup
from cyborgbackup.main.utils.common import parse_yaml_or_json


def validate_pem(data, min_keys=0, max_keys=None, min_certs=0, max_certs=None):
    """
    Validate the given PEM data is valid and contains the required numbers of
    keys and certificates.

    Return a list of PEM objects, where each object is a dict with the following
    keys:
      - 'all': The entire string for the PEM object including BEGIN/END lines.
      - 'type': The type of PEM object ('PRIVATE KEY' or 'CERTIFICATE').
      - 'data': The string inside the BEGIN/END lines.
      - 'b64': Key/certificate as a base64-encoded string.
      - 'bin': Key/certificate as bytes.
      - 'key_type': Only when type == 'PRIVATE KEY', one of 'rsa', 'dsa',
        'ecdsa', 'ed25519' or 'rsa1'.
      - 'key_enc': Only when type == 'PRIVATE KEY', boolean indicating if key is
        encrypted.
    """

    # Map the X in BEGIN X PRIVATE KEY to the key type (ssh-keygen -t).
    # Tower jobs using OPENSSH format private keys may still fail if the
    # system SSH implementation lacks support for this format.
    private_key_types = {
        'RSA': 'rsa',
        'DSA': 'dsa',
        'EC': 'ecdsa',
        'OPENSSH': 'ed25519',
        '': 'rsa1',
    }

    # Build regular expressions for matching each object in the PEM file.
    pem_obj_re = re.compile(
        r'^(?P<dashes>-{4,}) *BEGIN (?P<type>[A-Z ]+?) *(?P=dashes)' +
        r'\s*(?P<data>.+?)\s*' +
        r'(?P=dashes) *END (?P=type) *(?P=dashes)' +
        r'(?P<next>.*?)$', re.DOTALL
    )
    pem_obj_header_re = re.compile(r'^(.+?):\s*?(.+?)(\\??)$')

    pem_objects = []
    key_count, cert_count = 0, 0

    # Strip leading whitespaces at the start of the PEM data
    data = data.lstrip()

    while data:
        match = pem_obj_re.match(data)
        if not match:
            raise ValidationError(_('Invalid certificate or key: %s...') % data[:100])

        # The rest of the PEM data to process
        data = match.group('next').lstrip()

        # Check PEM object type, check key type if private key.
        pem_obj_info = {}
        pem_obj_info['all'] = match.group(0)
        pem_obj_info['type'] = pem_obj_type = match.group('type')
        if pem_obj_type.endswith('PRIVATE KEY'):
            key_count += 1
            pem_obj_info['type'] = 'PRIVATE KEY'
            key_type = pem_obj_type.replace('PRIVATE KEY', '').strip()
            try:
                pem_obj_info['key_type'] = private_key_types[key_type]
            except KeyError:
                raise ValidationError(_('Invalid private key: unsupported type "%s"') % key_type)
        elif pem_obj_type == 'CERTIFICATE':
            cert_count += 1
        else:
            raise ValidationError(_('Unsupported PEM object type: "%s"') % pem_obj_type)

        # Ensure that this PEM object is valid base64 data.
        pem_obj_info['data'] = match.group('data')
        base64_data = ''
        line_continues = False
        for line in pem_obj_info['data'].splitlines():
            line = line.strip()
            if not line:
                continue
            if line_continues:
                line_continues = line.endswith('\\')
                continue
            line_match = pem_obj_header_re.match(line)
            if line_match:
                line_continues = line.endswith('\\')
                continue
            base64_data += line
        try:
            decoded_data = base64.b64decode(base64_data)
            if not decoded_data:
                raise TypeError
            pem_obj_info['b64'] = base64_data
            pem_obj_info['bin'] = decoded_data
        except TypeError:
            raise ValidationError(_('Invalid base64-encoded data'))

        # If private key, check whether it is encrypted.
        if pem_obj_info.get('key_type', '') == 'ed25519':
            # See https://github.com/openssh/openssh-portable/blob/master/sshkey.c#L3218
            # Decoded key data starts with magic string (null-terminated), four byte
            # length field, followed by the ciphername -- if ciphername is anything
            # other than 'none' the key is encrypted.
            magic = 'openssh-key-v1\x00\x00\x00\x00\x04none'
            pem_obj_info['key_enc'] = not bool(pem_obj_info['bin'].startswith(magic))
        elif pem_obj_info.get('key_type', ''):
            pem_obj_info['key_enc'] = bool('ENCRYPTED' in pem_obj_info['data'])

        pem_objects.append(pem_obj_info)

    # Validate that the number of keys and certs provided are within the limits.
    kdict = dict(min=min_keys, max=max_keys, cnt=key_count)
    if key_count < min_keys:
        if min_keys == 1:
            if max_keys == min_keys:
                raise ValidationError(_('Exactly one private key is required.'))
            else:
                raise ValidationError(_('At least one private key is required.'))
        else:
            raise ValidationError(_('At least %(min)d private keys are required, only %(cnt)d provided.') % kdict)
    elif max_keys is not None and key_count > max_keys:
        if max_keys == 1:
            raise ValidationError(_('Only one private key is allowed, %(cnt)d provided.') % kdict)
        else:
            raise ValidationError(_('No more than %(max)d private keys are allowed, %(cnt)d provided.') % kdict)
    cdict = dict(min=min_certs, max=max_certs, cnt=cert_count)
    if cert_count < min_certs:
        if min_certs == 1:
            if max_certs == min_certs:
                raise ValidationError(_('Exactly one certificate is required.'))
            else:
                raise ValidationError(_('At least one certificate is required.'))
        else:
            raise ValidationError(_('At least %(min)d certificates are required, only %(cnt)d provided.') % cdict)
    elif max_certs is not None and cert_count > max_certs:
        if max_certs == 1:
            raise ValidationError(_('Only one certificate is allowed, %(cnt)d provided.') % cdict)
        else:
            raise ValidationError(_('No more than %(max)d certificates are allowed, %(cnt)d provided.') % cdict)

    return pem_objects


def validate_private_key(data):
    """
    Validate that data contains exactly one private key.
    """
    return validate_pem(data, min_keys=1, max_keys=1, max_certs=0)


def validate_certificate(data):
    """
    Validate that data contains one or more certificates. Adds BEGIN/END lines
    if necessary.
    """
    if 'BEGIN' not in data:
        data = "-----BEGIN CERTIFICATE-----\n{}".format(data)
    if 'END' not in data:
        data = "{}\n-----END CERTIFICATE-----\n".format(data)
    return validate_pem(data, max_keys=0, min_certs=1)


def validate_ssh_private_key(data):
    """
    Validate that data contains at least one private key and optionally
    certificates; should handle any valid options for ssh_private_key on a
    credential.
    """
    return validate_pem(data, min_keys=1)


def vars_validate_or_raise(vars_str):
    """
    Validate that fields like extra_vars or variables on resources like
    job templates, inventories, or hosts are either an acceptable
    blank string, or are valid JSON or YAML dict
    """
    try:
        parse_yaml_or_json(vars_str, silent_failure=False)
        return vars_str
    except ParseError as e:
        raise RestValidationError(str(e))

Read our documentation on viewing source code .

Navigation	Overlay
`t` Navigate files	`h` Toggle hits
`y` Change url to tip of branch	`m` Toggle misses
`b / v` Jump to prev/next hit line	`p` Toggle partial
`z / x` Jump to prev/next missed or partial line	`1..9` Toggle flags
`shift + o` Open current page in GitHub	`a` Toggle all on
`/ or ?` Show keyboard shortcuts dialog	`c` Toggle context lines or commits

1		# Python
2	6	import base64
3	6	import re
4
5		# Django
6	6	from django.utils.translation import ugettext_lazy as _
7	6	from django.core.exceptions import ValidationError
8
9		# REST framework
10	6	from rest_framework.serializers import ValidationError as RestValidationError
11	6	from rest_framework.exceptions import ParseError
12
13		# CyBorgBackup
14	6	from cyborgbackup.main.utils.common import parse_yaml_or_json
15
16
17	6	def validate_pem(data, min_keys=0, max_keys=None, min_certs=0, max_certs=None):
18		"""
19		Validate the given PEM data is valid and contains the required numbers of
20		keys and certificates.
21
22		Return a list of PEM objects, where each object is a dict with the following
23		keys:
24		- 'all': The entire string for the PEM object including BEGIN/END lines.
25		- 'type': The type of PEM object ('PRIVATE KEY' or 'CERTIFICATE').
26		- 'data': The string inside the BEGIN/END lines.
27		- 'b64': Key/certificate as a base64-encoded string.
28		- 'bin': Key/certificate as bytes.
29		- 'key_type': Only when type == 'PRIVATE KEY', one of 'rsa', 'dsa',
30		'ecdsa', 'ed25519' or 'rsa1'.
31		- 'key_enc': Only when type == 'PRIVATE KEY', boolean indicating if key is
32		encrypted.
33		"""
34
35		# Map the X in BEGIN X PRIVATE KEY to the key type (ssh-keygen -t).
36		# Tower jobs using OPENSSH format private keys may still fail if the
37		# system SSH implementation lacks support for this format.
38	0	private_key_types = {
39		'RSA': 'rsa',
40		'DSA': 'dsa',
41		'EC': 'ecdsa',
42		'OPENSSH': 'ed25519',
43		'': 'rsa1',
44		}
45
46		# Build regular expressions for matching each object in the PEM file.
47	0	pem_obj_re = re.compile(
48		r'^(?P<dashes>-{4,}) BEGIN (?P<type>[A-Z ]+?) (?P=dashes)' +
49		r'\s(?P<data>.+?)\s' +
50		r'(?P=dashes) END (?P=type) (?P=dashes)' +
51		r'(?P<next>.*?)$', re.DOTALL
52		)
53	0	pem_obj_header_re = re.compile(r'^(.+?):\s*?(.+?)(\\??)$')
54
55	0	pem_objects = []
56	0	key_count, cert_count = 0, 0
57
58		# Strip leading whitespaces at the start of the PEM data
59	0	data = data.lstrip()
60
61	0	while data:
62	0	match = pem_obj_re.match(data)
63	0	if not match:
64	0	raise ValidationError(_('Invalid certificate or key: %s...') % data[:100])
65
66		# The rest of the PEM data to process
67	0	data = match.group('next').lstrip()
68
69		# Check PEM object type, check key type if private key.
70	0	pem_obj_info = {}
71	0	pem_obj_info['all'] = match.group(0)
72	0	pem_obj_info['type'] = pem_obj_type = match.group('type')
73	0	if pem_obj_type.endswith('PRIVATE KEY'):
74	0	key_count += 1
75	0	pem_obj_info['type'] = 'PRIVATE KEY'
76	0	key_type = pem_obj_type.replace('PRIVATE KEY', '').strip()
77	0	try:
78	0	pem_obj_info['key_type'] = private_key_types[key_type]
79	0	except KeyError:
80	0	raise ValidationError(_('Invalid private key: unsupported type "%s"') % key_type)
81	0	elif pem_obj_type == 'CERTIFICATE':
82	0	cert_count += 1
83		else:
84	0	raise ValidationError(_('Unsupported PEM object type: "%s"') % pem_obj_type)
85
86		# Ensure that this PEM object is valid base64 data.
87	0	pem_obj_info['data'] = match.group('data')
88	0	base64_data = ''
89	0	line_continues = False
90	0	for line in pem_obj_info['data'].splitlines():
91	0	line = line.strip()
92	0	if not line:
93	0	continue
94	0	if line_continues:
95	0	line_continues = line.endswith('\\')
96	0	continue
97	0	line_match = pem_obj_header_re.match(line)
98	0	if line_match:
99	0	line_continues = line.endswith('\\')
100	0	continue
101	0	base64_data += line
102	0	try:
103	0	decoded_data = base64.b64decode(base64_data)
104	0	if not decoded_data:
105	0	raise TypeError
106	0	pem_obj_info['b64'] = base64_data
107	0	pem_obj_info['bin'] = decoded_data
108	0	except TypeError:
109	0	raise ValidationError(_('Invalid base64-encoded data'))
110
111		# If private key, check whether it is encrypted.
112	0	if pem_obj_info.get('key_type', '') == 'ed25519':
113		# See https://github.com/openssh/openssh-portable/blob/master/sshkey.c#L3218
114		# Decoded key data starts with magic string (null-terminated), four byte
115		# length field, followed by the ciphername -- if ciphername is anything
116		# other than 'none' the key is encrypted.
117	0	magic = 'openssh-key-v1\x00\x00\x00\x00\x04none'
118	0	pem_obj_info['key_enc'] = not bool(pem_obj_info['bin'].startswith(magic))
119	0	elif pem_obj_info.get('key_type', ''):
120	0	pem_obj_info['key_enc'] = bool('ENCRYPTED' in pem_obj_info['data'])
121
122	0	pem_objects.append(pem_obj_info)
123
124		# Validate that the number of keys and certs provided are within the limits.
125	0	kdict = dict(min=min_keys, max=max_keys, cnt=key_count)
126	0	if key_count < min_keys:
127	0	if min_keys == 1:
128	0	if max_keys == min_keys:
129	0	raise ValidationError(_('Exactly one private key is required.'))
130		else:
131	0	raise ValidationError(_('At least one private key is required.'))
132		else:
133	0	raise ValidationError(_('At least %(min)d private keys are required, only %(cnt)d provided.') % kdict)
134	0	elif max_keys is not None and key_count > max_keys:
135	0	if max_keys == 1:
136	0	raise ValidationError(_('Only one private key is allowed, %(cnt)d provided.') % kdict)
137		else:
138	0	raise ValidationError(_('No more than %(max)d private keys are allowed, %(cnt)d provided.') % kdict)
139	0	cdict = dict(min=min_certs, max=max_certs, cnt=cert_count)
140	0	if cert_count < min_certs:
141	0	if min_certs == 1:
142	0	if max_certs == min_certs:
143	0	raise ValidationError(_('Exactly one certificate is required.'))
144		else:
145	0	raise ValidationError(_('At least one certificate is required.'))
146		else:
147	0	raise ValidationError(_('At least %(min)d certificates are required, only %(cnt)d provided.') % cdict)
148	0	elif max_certs is not None and cert_count > max_certs:
149	0	if max_certs == 1:
150	0	raise ValidationError(_('Only one certificate is allowed, %(cnt)d provided.') % cdict)
151		else:
152	0	raise ValidationError(_('No more than %(max)d certificates are allowed, %(cnt)d provided.') % cdict)
153
154	0	return pem_objects
155
156
157	6	def validate_private_key(data):
158		"""
159		Validate that data contains exactly one private key.
160		"""
161	0	return validate_pem(data, min_keys=1, max_keys=1, max_certs=0)
162
163
164	6	def validate_certificate(data):
165		"""
166		Validate that data contains one or more certificates. Adds BEGIN/END lines
167		if necessary.
168		"""
169	0	if 'BEGIN' not in data:
170	0	data = "-----BEGIN CERTIFICATE-----\n{}".format(data)
171	0	if 'END' not in data:
172	0	data = "{}\n-----END CERTIFICATE-----\n".format(data)
173	0	return validate_pem(data, max_keys=0, min_certs=1)
174
175
176	6	def validate_ssh_private_key(data):
177		"""
178		Validate that data contains at least one private key and optionally
179		certificates; should handle any valid options for ssh_private_key on a
180		credential.
181		"""
182	0	return validate_pem(data, min_keys=1)
183
184
185	6	def vars_validate_or_raise(vars_str):
186		"""
187		Validate that fields like extra_vars or variables on resources like
188		job templates, inventories, or hosts are either an acceptable
189		blank string, or are valid JSON or YAML dict
190		"""
191	0	try:
192	0	parse_yaml_or_json(vars_str, silent_failure=False)
193	0	return vars_str
194	0	except ParseError as e:
195	0	raise RestValidationError(str(e))