1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
|
# (c) 2012-2014, Michael DeHaan <michael.dehaan@gmail.com>
#
# This file is part of Ansible
#
# Ansible is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# Ansible is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with Ansible. If not, see <http://www.gnu.org/licenses/>.
# Make coding more python3-ish
from __future__ import (absolute_import, division, print_function)
__metaclass__ = type
import yaml
from ansible.compat.six import text_type
from ansible.module_utils._text import to_bytes
class AnsibleBaseYAMLObject(object):
'''
the base class used to sub-class python built-in objects
so that we can add attributes to them during yaml parsing
'''
_data_source = None
_line_number = 0
_column_number = 0
def _get_ansible_position(self):
return (self._data_source, self._line_number, self._column_number)
def _set_ansible_position(self, obj):
try:
(src, line, col) = obj
except (TypeError, ValueError):
raise AssertionError(
'ansible_pos can only be set with a tuple/list '
'of three values: source, line number, column number'
)
self._data_source = src
self._line_number = line
self._column_number = col
ansible_pos = property(_get_ansible_position, _set_ansible_position)
class AnsibleMapping(AnsibleBaseYAMLObject, dict):
''' sub class for dictionaries '''
pass
class AnsibleUnicode(AnsibleBaseYAMLObject, text_type):
''' sub class for unicode objects '''
pass
class AnsibleSequence(AnsibleBaseYAMLObject, list):
''' sub class for lists '''
pass
# Unicode like object that is not evaluated (decrypted) until it needs to be
# TODO: is there a reason these objects are subclasses for YAMLObject?
class AnsibleVaultEncryptedUnicode(yaml.YAMLObject, AnsibleUnicode):
__UNSAFE__ = True
__ENCRYPTED__ = True
yaml_tag = u'!vault-encrypted'
@classmethod
def from_plaintext(cls, seq, vault):
if not vault:
raise vault.AnsibleVaultError('Error creating AnsibleVaultEncryptedUnicode, invalid vault (%s) provided' % vault)
ciphertext = vault.encrypt(seq)
avu = cls(ciphertext)
avu.vault = vault
return avu
def __init__(self, ciphertext):
'''A AnsibleUnicode with a Vault attribute that can decrypt it.
ciphertext is a byte string (str on PY2, bytestring on PY3).
The .data atttribute is a property that returns the decrypted plaintext
of the ciphertext as a PY2 unicode or PY3 string object.
'''
super(AnsibleVaultEncryptedUnicode, self).__init__()
# after construction, calling code has to set the .vault attribute to a vaultlib object
self.vault = None
self._ciphertext = to_bytes(ciphertext)
@property
def data(self):
if not self.vault:
# FIXME: raise exception?
return self._ciphertext
return self.vault.decrypt(self._ciphertext).decode()
@data.setter
def data(self, value):
self._ciphertext = value
def __repr__(self):
return 'AnsibleVaultEncryptedUnicode(%s)' % self._ciphertext
# Compare a regular str/text_type with the decrypted hypertext
def __eq__(self, other):
if self.vault:
return other == self.data
return False
def __hash__(self):
return id(self)
def __ne__(self, other):
if self.vault:
return other != self.data
return True
def __str__(self):
return str(self.data)
def __unicode__(self):
return unicode(self.data)
def encode(self, encoding=None, errors=None):
return self.data.encode(encoding, errors)
|
