TermsHost.__init__() - Code Metrics - Inspection of "change branch back to master" - peshay/tpmstore - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 898e86...574dfa )

by Andreas

created 2017-07-09 08:27 UTC

TermsHost.init() A

↳ Parent: TermsHost

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	3
Bugs	1	Features	0

Metric	Value
cc	2
c	3
b	1
f	0
dl	0
loc	11
rs	9.4285

#
# tpmstore - TeamPasswordManager lookup plugin for Ansible.
# Copyright (C) 2017 Andreas Hubert
# See LICENSE.txt for licensing details
#
# File: tpmstore.py
#

from ansible.errors import AnsibleError, AnsibleParserError
from ansible.plugins.lookup import LookupBase
import tpm
"""
DOCUMENTATION:
    lookup: tpmstore
    version_added: "2.4"
    short_description: returns password from TeamPasswordManager
    description:
        - Queries TeamPasswordManager API to get a password from an entry.
          Entries in TeamPasswordManager can also be created or updated.

    options:
        tpmurl:
            description:
                - URL to TeamPasswordManager API. Should always be first parameter.
            required: True
        tpmuser:
            description:
                - User to authenticate against TeamPasswordManager API. Should always be second parameter.
            required: True
        tpmpass:
            description:
                - Password to authenticate against TeamPasswordManager API. Should always be third parameter.
            required: True
        search:
            description:
                - Searchtstring to use for the TeamPasswordManager search.
            required: If 'name' is not set.
            default: 'name:[name]'
        name:
            description:
                - Name of the entry in TeamPasswordManager. Will search for exact match.
            required: If 'search' is not set.
        return_value:
            description:
                - Which fields from found entries should be returned.
            required: False
            default: password
        create:
            description:
                - If False the plugin will only query for a password.
                  If True it will update an existing entry or create a new entry if it does not exists in TeamPasswordManager,
                  in this case project_id will be required.
            possible values: True, False
            default: False
        reason:
            description:
                - If an entry is locked, an unlock reason is mandatory.
    options if create=True:
        project_id:
            description:
                - If a complete new entry is created, we need to assign it to an existing project in TeamPasswordManager.
            required: Only if create=True AND no entry by "name" already exists.
        password:
            description:
                - Will update or set the field "password" for the TeamPasswordManager entry.
                  If set to "random" a new random password will be generated, updated to TeamPasswordManager and returned.
        username:
            description:
                - Wil update or set the field "username" for the TeamPasswordManager entry.
        access_info:
            description:
                - Wil update or set the field "access_info" for the TeamPasswordManager entry.
        tags:
            description:
                - Wil update or set the field "tags" for the TeamPasswordManager entry.
        email:
            description:
                - Wil update or set the field "email" for the TeamPasswordManager entry.
        expiry_date:
            description:
                - Wil update or set the field "expiry_date" for the TeamPasswordManager entry.
        notes:
            description:
                - Wil update or set the field "notes" for the TeamPasswordManager entry.
EXAMPLES:
  vars_prompt:
    - name: "tpmuser"
      prompt: "what is your TeamPasswordManager username?"
      private: no
    - name: "tpmpass"
      prompt: "what is your TeamPasswordManager password?"
      private: yes
  vars:
     tpmurl:   "https://MyTpmHost.example.com"
     retrieve_password: "{{ lookup('tpmstore', tpmurl, tpmuser, tpmpass, 'name=An existing entry name') }}"
     retrieve_username: "{{ lookup('tpmstore', tpmurl, tpmuser, tpmpass, 'name=An existing entry name', 'return_value=username')}}"
     search_by_tags: "{{ lookup('tpmstore', tpmurl, tpmuser, tpmpass, 'search=tags:sshhost') }}"
     retrieve_locked_password: "{{ lookup('tpmstore', tpmurl, tpmuser, tpmpass, 'name=An existing and locked entry name', 'reason=For Auto Deploy by Ansible') }}"
     newrandom_password: "{{ lookup('tpmstore', tpmurl, tpmuser, tpmpass, 'name=An existing entry name', 'create=True', 'password=random') }}"
     updatemore_values: "{{ lookup('tpmstore', tpmurl, tpmuser, tpmpass, 'name=An existing entry name', 'create=True', 'password=random', 'username=root', 'access_info=ssh://root@host', 'tags=root,ssh,aws,cloud', 'notes=Created by Ansible') }}"
     completenew_entry: "{{ lookup('tpmstore', tpmurl, tpmuser, tpmpass, 'name=An existing entry name', 'create=True', 'project_id=4', 'password=random', 'username=root', 'access_info=ssh://root@host', 'tags=root,ssh,aws,cloud', 'notes=Created by Ansible') }}"


RETURN:
  _list:
    description:
      - list containing the queried or created password
    type: lists
"""

try:
    from __main__ import display
except ImportError:
    from ansible.utils.display import Display
    display = Display()

class TermsHost(object):
    
    def __init__(self, terms):
        # We need at least 4 parameters: api-url, api-user, api-password, entry name
        if len(terms) < 4:
            raise AnsibleError("At least 4 arguments required.")
        # Fill the mandatory values
        self.tpmurl=terms.pop(0)
        self.tpmuser=terms.pop(0)
        self.tpmpass=terms.pop(0)
        self.work_on_terms(terms)
        self.verify_values()
        self.match = self.initiate_search()
    
    def verify_values(self):
        """Verify the correctness of all the values."""
        # verify if either search or name is set
        if not hasattr(self, 'name') and not hasattr(self, 'search'):
            raise AnsibleError('Either "name" or "search" have to be set.')
            
    def work_on_terms(self, terms):
        """Collect all the terms."""
        self.create = False
        self.new_entry = {}
        for term in terms:
            if "=" in term:
                (key, value) = term.split("=")
                # entry name is mandatory
                if key == "name":
                    # get entry
                    self.name = value
                    self.new_entry.update({'name': self.name})
                if key == 'search':
                    self.search = value
                if key == 'return_value':
                    self.return_value = value
                # if not just lookup, but also create/update an entry
                if key == "create":
                    if value == "True":
                        self.create = True
                    elif value == "False":
                        self.create = False
                    else:
                        raise AnsibleError("create can only be True or False and not: {}".format(value))
                # optional parameters for create/update of an entry
                if key == "password":
                    self.password = value
                    self.new_entry.update({'password': self.password})
                if key == "username":
                    self.username = value
                    self.new_entry.update({'username': self.username})
                if key == "access_info":
                    self.access_info = value
                    self.new_entry.update({'access_info': self.access_info})
                if key == "tags":
                    self.tags = value
                    self.new_entry.update({'tags': self.tags})
                if key == "email":
                    self.email = value
                    self.new_entry.update({'email': self.email})
                if key == "expiry_date":
                    self.expiry_date = value
                    self.new_entry.update({'expiry_date': self.expiry_date})
                if key == "notes":
                    self.notes = value
                    self.new_entry.update({'notes': self.notes})
                if key == "reason":
                    self.unlock_reason = value
                # project_id is mandatory if no entry exists and create == True
                if key == "project_id":
                    self.project_id = value
                    self.new_entry.update({'project_id': self.project_id})

    def initiate_search(self):
        # format the search to get an exact result for name
        if hasattr(self, 'search'):
            search = self.search
        else:
            search = "name:[{}]".format(self.name)
        # set default return_value to 'password'
        if not hasattr(self, 'return_value'):
            self.return_value = 'password'
            
        try:
            if hasattr(self, "unlock_reason"):
                self.tpmconn = tpm.TpmApiv4(self.tpmurl, username=self.tpmuser, password=self.tpmpass, unlock_reason=self.unlock_reason)
            else:
                self.tpmconn = tpm.TpmApiv4(self.tpmurl, username=self.tpmuser, password=self.tpmpass)
            match = self.tpmconn.list_passwords_search(search)
        except tpm.TpmApiv4.ConfigError as e:
            raise AnsibleError("First argument has to be a valid URL to TeamPasswordManager API: {}".format(self.tpmurl))
        except tpm.TPMException as e:
            raise AnsibleError(e)
        return match


class LookupModule(LookupBase):

    def run(self, terms, variables=None, **kwargs):
        ret = []
        th = TermsHost(terms)

        # If there are no entries and we should create
        if len(th.match) < 1 and th.create == True:
            display.display("No entry found, will create: {}".format(th.name))
            if hasattr(th, "project_id"):
                if hasattr(th, "password"):
                    if th.password == "random":
                        new_password = th.tpmconn.generate_password().get("password")
                        th.new_entry.update({'password': new_password})
                        th.password = new_password
                try:
                    newid = th.tpmconn.create_password(th.new_entry)
                    display.display("Created new entry with ID: {}".format(newid.get('id')))
                    ret = [th.password]
                except tpm.TPMException as e:
                    raise AnsibleError(e)
            else:
                raise AnsibleError("To create a complete new entry, project_id is mandatory.")
        elif len(th.match) < 1 and th.create == False:
            raise AnsibleError("Found no match for: {}".format(th.name))
        elif len(th.match) > 1:
            raise AnsibleError("Found more then one match for the entry, please be more specific: {}".format(th.name))
        elif th.create == True:
            result = th.tpmconn.show_password(th.match[0])
            display.display('Will update entry "{}" with ID "{}"'.format(result.get("name"), result.get("id")))
            if hasattr(th, "password"):
                if th.password == "random":
                    new_password = th.tpmconn.generate_password().get("password")
                    th.new_entry.update({'password': new_password})
                    th.password = new_password
            try:
                th.tpmconn.update_password(result.get("id"),th.new_entry)
                ret = [th.password]
            except tpm.TPMException as e:
                raise AnsibleError(e)
        else:
            result = th.tpmconn.show_password(th.match[0].get("id"))
            ret = [result.get(th.return_value)]

        return ret


1			#
2			# tpmstore - TeamPasswordManager lookup plugin for Ansible.
3			# Copyright (C) 2017 Andreas Hubert
4			# See LICENSE.txt for licensing details
5			#
6			# File: tpmstore.py
7			#
8
9			from ansible.errors import AnsibleError, AnsibleParserError
10			from ansible.plugins.lookup import LookupBase
11			import tpm
12			"""
13			DOCUMENTATION:
14			lookup: tpmstore
15			version_added: "2.4"
16			short_description: returns password from TeamPasswordManager
17			description:
18			- Queries TeamPasswordManager API to get a password from an entry.
19			Entries in TeamPasswordManager can also be created or updated.
20
21			options:
22			tpmurl:
23			description:
24			- URL to TeamPasswordManager API. Should always be first parameter.
25			required: True
26			tpmuser:
27			description:
28			- User to authenticate against TeamPasswordManager API. Should always be second parameter.
29			required: True
30			tpmpass:
31			description:
32			- Password to authenticate against TeamPasswordManager API. Should always be third parameter.
33			required: True
34			search:
35			description:
36			- Searchtstring to use for the TeamPasswordManager search.
37			required: If 'name' is not set.
38			default: 'name:[name]'
39			name:
40			description:
41			- Name of the entry in TeamPasswordManager. Will search for exact match.
42			required: If 'search' is not set.
43			return_value:
44			description:
45			- Which fields from found entries should be returned.
46			required: False
47			default: password
48			create:
49			description:
50			- If False the plugin will only query for a password.
51			If True it will update an existing entry or create a new entry if it does not exists in TeamPasswordManager,
52			in this case project_id will be required.
53			possible values: True, False
54			default: False
55			reason:
56			description:
57			- If an entry is locked, an unlock reason is mandatory.
58			options if create=True:
59			project_id:
60			description:
61			- If a complete new entry is created, we need to assign it to an existing project in TeamPasswordManager.
62			required: Only if create=True AND no entry by "name" already exists.
63			password:
64			description:
65			- Will update or set the field "password" for the TeamPasswordManager entry.
66			If set to "random" a new random password will be generated, updated to TeamPasswordManager and returned.
67			username:
68			description:
69			- Wil update or set the field "username" for the TeamPasswordManager entry.
70			access_info:
71			description:
72			- Wil update or set the field "access_info" for the TeamPasswordManager entry.
73			tags:
74			description:
75			- Wil update or set the field "tags" for the TeamPasswordManager entry.
76			email:
77			description:
78			- Wil update or set the field "email" for the TeamPasswordManager entry.
79			expiry_date:
80			description:
81			- Wil update or set the field "expiry_date" for the TeamPasswordManager entry.
82			notes:
83			description:
84			- Wil update or set the field "notes" for the TeamPasswordManager entry.
85			EXAMPLES:
86			vars_prompt:
87			- name: "tpmuser"
88			prompt: "what is your TeamPasswordManager username?"
89			private: no
90			- name: "tpmpass"
91			prompt: "what is your TeamPasswordManager password?"
92			private: yes
93			vars:
94			tpmurl: "https://MyTpmHost.example.com"
95			retrieve_password: "{{ lookup('tpmstore', tpmurl, tpmuser, tpmpass, 'name=An existing entry name') }}"
96			retrieve_username: "{{ lookup('tpmstore', tpmurl, tpmuser, tpmpass, 'name=An existing entry name', 'return_value=username')}}"
97			search_by_tags: "{{ lookup('tpmstore', tpmurl, tpmuser, tpmpass, 'search=tags:sshhost') }}"
98			retrieve_locked_password: "{{ lookup('tpmstore', tpmurl, tpmuser, tpmpass, 'name=An existing and locked entry name', 'reason=For Auto Deploy by Ansible') }}"
99			newrandom_password: "{{ lookup('tpmstore', tpmurl, tpmuser, tpmpass, 'name=An existing entry name', 'create=True', 'password=random') }}"
100			updatemore_values: "{{ lookup('tpmstore', tpmurl, tpmuser, tpmpass, 'name=An existing entry name', 'create=True', 'password=random', 'username=root', 'access_info=ssh://root@host', 'tags=root,ssh,aws,cloud', 'notes=Created by Ansible') }}"
101			completenew_entry: "{{ lookup('tpmstore', tpmurl, tpmuser, tpmpass, 'name=An existing entry name', 'create=True', 'project_id=4', 'password=random', 'username=root', 'access_info=ssh://root@host', 'tags=root,ssh,aws,cloud', 'notes=Created by Ansible') }}"
102
103
104			RETURN:
105			_list:
106			description:
107			- list containing the queried or created password
108			type: lists
109			"""
110
111			try:
112			from __main__ import display
113			except ImportError:
114			from ansible.utils.display import Display
115			display = Display()
116
117			class TermsHost(object):
118
119			def __init__(self, terms):
120			# We need at least 4 parameters: api-url, api-user, api-password, entry name
121			if len(terms) < 4:
122			raise AnsibleError("At least 4 arguments required.")
123			# Fill the mandatory values
124			self.tpmurl=terms.pop(0)
125			self.tpmuser=terms.pop(0)
126			self.tpmpass=terms.pop(0)
127			self.work_on_terms(terms)
128			self.verify_values()
129			self.match = self.initiate_search()
130
131			def verify_values(self):
132			"""Verify the correctness of all the values."""
133			# verify if either search or name is set
134			if not hasattr(self, 'name') and not hasattr(self, 'search'):
135			raise AnsibleError('Either "name" or "search" have to be set.')
136
137			def work_on_terms(self, terms):
138			"""Collect all the terms."""
139			self.create = False
140			self.new_entry = {}
141			for term in terms:
142			if "=" in term:
143			(key, value) = term.split("=")
144			# entry name is mandatory
145			if key == "name":
146			# get entry
147			self.name = value
148			self.new_entry.update({'name': self.name})
149			if key == 'search':
150			self.search = value
151			if key == 'return_value':
152			self.return_value = value
153			# if not just lookup, but also create/update an entry
154			if key == "create":
155			if value == "True":
156			self.create = True
157			elif value == "False":
158			self.create = False
159			else:
160			raise AnsibleError("create can only be True or False and not: {}".format(value))
161			# optional parameters for create/update of an entry
162			if key == "password":
163			self.password = value
164			self.new_entry.update({'password': self.password})
165			if key == "username":
166			self.username = value
167			self.new_entry.update({'username': self.username})
168			if key == "access_info":
169			self.access_info = value
170			self.new_entry.update({'access_info': self.access_info})
171			if key == "tags":
172			self.tags = value
173			self.new_entry.update({'tags': self.tags})
174			if key == "email":
175			self.email = value
176			self.new_entry.update({'email': self.email})
177			if key == "expiry_date":
178			self.expiry_date = value
179			self.new_entry.update({'expiry_date': self.expiry_date})
180			if key == "notes":
181			self.notes = value
182			self.new_entry.update({'notes': self.notes})
183			if key == "reason":
184			self.unlock_reason = value
185			# project_id is mandatory if no entry exists and create == True
186			if key == "project_id":
187			self.project_id = value
188			self.new_entry.update({'project_id': self.project_id})
189
190			def initiate_search(self):
191			# format the search to get an exact result for name
192			if hasattr(self, 'search'):
193			search = self.search
194			else:
195			search = "name:[{}]".format(self.name)
196			# set default return_value to 'password'
197			if not hasattr(self, 'return_value'):
198			self.return_value = 'password'
199
200			try:
201			if hasattr(self, "unlock_reason"):
202			self.tpmconn = tpm.TpmApiv4(self.tpmurl, username=self.tpmuser, password=self.tpmpass, unlock_reason=self.unlock_reason)
203			else:
204			self.tpmconn = tpm.TpmApiv4(self.tpmurl, username=self.tpmuser, password=self.tpmpass)
205			match = self.tpmconn.list_passwords_search(search)
206			except tpm.TpmApiv4.ConfigError as e:
207			raise AnsibleError("First argument has to be a valid URL to TeamPasswordManager API: {}".format(self.tpmurl))
208			except tpm.TPMException as e:
209			raise AnsibleError(e)
210			return match
211
212
213			class LookupModule(LookupBase):
214
215			def run(self, terms, variables=None, **kwargs):
216			ret = []
217			th = TermsHost(terms)
218
219			# If there are no entries and we should create
220			if len(th.match) < 1 and th.create == True:
221			display.display("No entry found, will create: {}".format(th.name))
222			if hasattr(th, "project_id"):
223			if hasattr(th, "password"):
224			if th.password == "random":
225			new_password = th.tpmconn.generate_password().get("password")
226			th.new_entry.update({'password': new_password})
227			th.password = new_password
228			try:
229			newid = th.tpmconn.create_password(th.new_entry)
230			display.display("Created new entry with ID: {}".format(newid.get('id')))
231			ret = [th.password]
232			except tpm.TPMException as e:
233			raise AnsibleError(e)
234			else:
235			raise AnsibleError("To create a complete new entry, project_id is mandatory.")
236			elif len(th.match) < 1 and th.create == False:
237			raise AnsibleError("Found no match for: {}".format(th.name))
238			elif len(th.match) > 1:
239			raise AnsibleError("Found more then one match for the entry, please be more specific: {}".format(th.name))
240			elif th.create == True:
241			result = th.tpmconn.show_password(th.match[0])
242			display.display('Will update entry "{}" with ID "{}"'.format(result.get("name"), result.get("id")))
243			if hasattr(th, "password"):
244			if th.password == "random":
245			new_password = th.tpmconn.generate_password().get("password")
246			th.new_entry.update({'password': new_password})
247			th.password = new_password
248			try:
249			th.tpmconn.update_password(result.get("id"),th.new_entry)
250			ret = [th.password]
251			except tpm.TPMException as e:
252			raise AnsibleError(e)
253			else:
254			result = th.tpmconn.show_password(th.match[0].get("id"))
255			ret = [result.get(th.return_value)]
256
257			return ret
258

peshay / tpmstore

Push — master ( 898e86...574dfa )

TermsHost.__init__() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

TermsHost.init() A