gvmtools.parser.CliParser._add_subparsers() - Code Metrics - Inspection of "Use a common parser for both cli tools" - greenbone/gvm-tools - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#140)

unknown

created 2018-12-13 14:54 UTC

gvmtools.parser.CliParser._add_subparsers() B

↳ Parent: gvmtools.parser

Complexity

Conditions

Size

Total Lines	65
Code Lines	54

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	54
nop	1
dl	0
loc	65
rs	8.5054
c	0
b	0
f	0

How to fix Long Method

# -*- coding: utf-8 -*-
# Copyright (C) 2018 Greenbone Networks GmbH
#
# SPDX-License-Identifier: GPL-3.0-or-later
#
# This program 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.
#
# This program 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 this program.  If not, see <http://www.gnu.org/licenses/>.

"""Command Line Interface Parser
"""

import argparse
import configparser
import logging
import os

from gvm import get_version as get_gvm_version
from gvm.connections import (DEFAULT_UNIX_SOCKET_PATH,
                             DEFAULT_TIMEOUT,
                             DEFAULT_GVM_PORT,
                             SSHConnection,
                             TLSConnection,
                             UnixSocketConnection)

from gvmtools import get_version

logger = logging.getLogger(__name__)

__version__ = get_version()
__api_version__ = get_gvm_version()

DEFAULT_CONFIG_PATH = '~/.config/gvm-tools.conf'

PROTOCOL_OSP = 'OSP'
PROTOCOL_GMP = 'GMP'
DEFAULT_PROTOCOL = PROTOCOL_GMP


class CliParser:

    def __init__(self, description, logfilename):
        root_parser = argparse.ArgumentParser(
            description=description,
            formatter_class=argparse.RawTextHelpFormatter,
            add_help=False)

        root_parser.add_argument(
            '-c', '--config', nargs='?', default=DEFAULT_CONFIG_PATH,
            help='Configuration file path (default: %(default)s)')
        root_parser.add_argument(
            '--log', nargs='?', dest='loglevel', const='INFO',
            choices=['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'],
            help='Activate logging (default level: %(default)s)')

        args_before, _ = root_parser.parse_known_args()

        if args_before.loglevel is not None:
            level = logging.getLevelName(args_before.loglevel)
            logging.basicConfig(filename=logfilename, level=level)

        root_parser.add_argument(
            '--timeout', required=False, default=DEFAULT_TIMEOUT, type=int,
            help='Response timeout in seconds, or -1 to wait '
                 'indefinitely (default: %(default)s)')
        root_parser.add_argument(
            '--gmp-username',
            help='Username for GMP service (default: %(default)r)')
        root_parser.add_argument(
            '--gmp-password',
            help='Password for GMP service (default: %(default)r)')
        root_parser.add_argument(
            '-V', '--version', action='version',
            version='%(prog)s {version} (API version {apiversion})'.format(
                version=__version__, apiversion=__api_version__),
            help='Show version information and exit')

        parser = argparse.ArgumentParser(parents=[root_parser])

        subparsers = parser.add_subparsers(
            metavar='CONNECTION_TYPE',
            title='connections',
            description='valid connection types',
            help="Connection type to use",
        )
        subparsers.required = True
        subparsers.dest = 'connection_type'

        self._config = self._load_config(args_before.config)

        self._parser = parser
        self._root_parser = root_parser
        self._subparsers = subparsers

    def parse_args(self):
        self._add_subparsers()

        self._parser.set_defaults(
            gmp_username=self._config.get('gmp', 'username', fallback=''),
            gmp_password=self._config.get('gmp', 'password', fallback=''),
            **self._config.defaults()
        )

        args = self._parser.parse_args()

        # If timeout value is -1, then the socket should have no timeout
        if args.timeout == -1:
            args.timeout = None


        logging.debug('Parsed arguments %r', args)

        return args

    def add_argument(self, *args, **kwargs):
        self._parser.add_argument(*args, **kwargs)
        self._root_parser.add_argument(*args, **kwargs)

    def _load_config(self, configfile):
        config = configparser.ConfigParser(default_section='main')

        try:
            path = os.path.expanduser(configfile)
            config.read(path)
        except Exception as e: # pylint: disable=broad-except
            raise RuntimeError(
                'Error while parsing config file {config}. Error was '
                '{message}'.format(config=configfile, message=e))

        logger.debug('Loaded config %s', configfile)
        return config

    def _add_subparsers(self):
        parser_ssh = self._subparsers.add_parser(
            'ssh', help='Use SSH to connect to service',
            parents=[self._root_parser])

        parser_ssh.add_argument('--hostname', required=True,
                                help='Hostname or IP address')
        parser_ssh.add_argument('--port', required=False,
                                help='SSH port (default: %(default)s)')
        parser_ssh.add_argument('--ssh-username',
                                help='SSH username (default: %(default)r)')
        parser_ssh.add_argument('--ssh-password',
                                help='SSH password (default: %(default)r)')

        parser_ssh.set_defaults(
            port=int(self._config.get('ssh', 'port', fallback=22)),
            ssh_username=self._config.get('ssh', 'username', fallback='gmp'),
            ssh_password=self._config.get('ssh', 'password', fallback='gmp'),
        )

        parser_tls = self._subparsers.add_parser(
            'tls', help='Use TLS secured connection to connect to service',
            parents=[self._root_parser])
        parser_tls.add_argument('--hostname', required=True,
                                help='Hostname or IP address')
        parser_tls.add_argument('--port', required=False,
                                help='GMP/OSP port (default: %(default)s)')
        parser_tls.add_argument(
            '--certfile', required=False,
            help='Path to the certificate file for client authentication. '
                 '(default: %(default)s)')
        parser_tls.add_argument(
            '--keyfile', required=False,
            help='Path to key file for client authentication. '
                 '(default: %(default)s)')
        parser_tls.add_argument(
            '--cafile', required=False,
            help='Path to CA certificate for server authentication. '
                 '(default: %(default)s)')
        parser_tls.add_argument(
            '--no-credentials', required=False, default=False,
            help='Use only certificates for authentication')
        parser_tls.set_defaults(
            port=int(self._config.get(
                'tls', 'port', fallback=DEFAULT_GVM_PORT)),
            certfile=self._config.get('tls', 'certfile', fallback=None),
            keyfile=self._config.get('tls', 'keyfile', fallback=None),
            cafile=self._config.get('tls', 'cafile', fallback=None),
        )

        parser_socket = self._subparsers.add_parser(
            'socket', help='Use UNIX Domain socket to connect to service',
            parents=[self._root_parser])

        socketpath_group = parser_socket.add_mutually_exclusive_group()
        socketpath_group.add_argument(
            '--sockpath', nargs='?', default=None,
            help='Deprecated, use --socketpath instead')
        socketpath_group.add_argument(
            '--socketpath', nargs='?',
            help='Path to UNIX Domain socket (default: %(default)s)')

        parser_socket.set_defaults(
            socketpath=self._config.get(
                'unixsocket', 'socketpath', fallback=DEFAULT_UNIX_SOCKET_PATH),
        )

    def add_protocol_argument(self):
        self.add_argument(
            '--protocol', required=False, default=DEFAULT_PROTOCOL,
            choices=[PROTOCOL_GMP, PROTOCOL_OSP],
            help='Service protocol to use (default: %(default)s)')


def create_parser(description, logfilename):
    return CliParser(description, logfilename)


def create_connection(connection_type, socketpath=None, timeout=None,
                      hostname=None, port=None, certfile=None, keyfile=None,
                      cafile=None, ssh_username=None, ssh_password=None,
                      **kwargs):
    if 'socket' in connection_type:
        return UnixSocketConnection(
            timeout=timeout,
            path=socketpath
        )

    if 'tls' in connection_type:
        return TLSConnection(
            timeout=timeout,
            hostname=hostname,
            port=port,
            certfile=certfile,
            keyfile=keyfile,
            cafile=cafile,
        )

    return SSHConnection(
        timeout=timeout,
        hostname=hostname,
        port=port,
        username=ssh_username,
        password=ssh_password
    )


1			# -- coding: utf-8 --
2			# Copyright (C) 2018 Greenbone Networks GmbH
3			#
4			# SPDX-License-Identifier: GPL-3.0-or-later
5			#
6			# This program is free software: you can redistribute it and/or modify
7			# it under the terms of the GNU General Public License as published by
8			# the Free Software Foundation, either version 3 of the License, or
9			# (at your option) any later version.
10			#
11			# This program is distributed in the hope that it will be useful,
12			# but WITHOUT ANY WARRANTY; without even the implied warranty of
13			# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
14			# GNU General Public License for more details.
15			#
16			# You should have received a copy of the GNU General Public License
17			# along with this program. If not, see <http://www.gnu.org/licenses/>.
18
19			"""Command Line Interface Parser
20			"""
21
22			import argparse
23			import configparser
24			import logging
25			import os
26
27			from gvm import get_version as get_gvm_version
28			from gvm.connections import (DEFAULT_UNIX_SOCKET_PATH,
29			DEFAULT_TIMEOUT,
30			DEFAULT_GVM_PORT,
31			SSHConnection,
32			TLSConnection,
33			UnixSocketConnection)
34
35			from gvmtools import get_version
36
37			logger = logging.getLogger(__name__)
38
39			__version__ = get_version()
40			__api_version__ = get_gvm_version()
41
42			DEFAULT_CONFIG_PATH = '~/.config/gvm-tools.conf'
43
44			PROTOCOL_OSP = 'OSP'
45			PROTOCOL_GMP = 'GMP'
46			DEFAULT_PROTOCOL = PROTOCOL_GMP
47
48
49			class CliParser:
50
51			def __init__(self, description, logfilename):
52			root_parser = argparse.ArgumentParser(
53			description=description,
54			formatter_class=argparse.RawTextHelpFormatter,
55			add_help=False)
56
57			root_parser.add_argument(
58			'-c', '--config', nargs='?', default=DEFAULT_CONFIG_PATH,
59			help='Configuration file path (default: %(default)s)')
60			root_parser.add_argument(
61			'--log', nargs='?', dest='loglevel', const='INFO',
62			choices=['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'],
63			help='Activate logging (default level: %(default)s)')
64
65			args_before, _ = root_parser.parse_known_args()
66
67			if args_before.loglevel is not None:
68			level = logging.getLevelName(args_before.loglevel)
69			logging.basicConfig(filename=logfilename, level=level)
70
71			root_parser.add_argument(
72			'--timeout', required=False, default=DEFAULT_TIMEOUT, type=int,
73			help='Response timeout in seconds, or -1 to wait '
74			'indefinitely (default: %(default)s)')
75			root_parser.add_argument(
76			'--gmp-username',
77			help='Username for GMP service (default: %(default)r)')
78			root_parser.add_argument(
79			'--gmp-password',
80			help='Password for GMP service (default: %(default)r)')
81			root_parser.add_argument(
82			'-V', '--version', action='version',
83			version='%(prog)s {version} (API version {apiversion})'.format(
84			version=__version__, apiversion=__api_version__),
85			help='Show version information and exit')
86
87			parser = argparse.ArgumentParser(parents=[root_parser])
88
89			subparsers = parser.add_subparsers(
90			metavar='CONNECTION_TYPE',
91			title='connections',
92			description='valid connection types',
93			help="Connection type to use",
94			)
95			subparsers.required = True
96			subparsers.dest = 'connection_type'
97
98			self._config = self._load_config(args_before.config)
99
100			self._parser = parser
101			self._root_parser = root_parser
102			self._subparsers = subparsers
103
104			def parse_args(self):
105			self._add_subparsers()
106
107			self._parser.set_defaults(
108			gmp_username=self._config.get('gmp', 'username', fallback=''),
109			gmp_password=self._config.get('gmp', 'password', fallback=''),
110			**self._config.defaults()
111			)
112
113			args = self._parser.parse_args()
114
115			# If timeout value is -1, then the socket should have no timeout
116			if args.timeout == -1:
117			args.timeout = None
118
119
120			logging.debug('Parsed arguments %r', args)
121
122			return args
123
124			def add_argument(self, args, *kwargs):
125			self._parser.add_argument(args, *kwargs)
126			self._root_parser.add_argument(args, *kwargs)
127
128			def _load_config(self, configfile):
129			config = configparser.ConfigParser(default_section='main')
130
131			try:
132			path = os.path.expanduser(configfile)
133			config.read(path)
134			except Exception as e: # pylint: disable=broad-except
135			raise RuntimeError(
136			'Error while parsing config file {config}. Error was '
137			'{message}'.format(config=configfile, message=e))
138
139			logger.debug('Loaded config %s', configfile)
140			return config
141
142			def _add_subparsers(self):
143			parser_ssh = self._subparsers.add_parser(
144			'ssh', help='Use SSH to connect to service',
145			parents=[self._root_parser])
146
147			parser_ssh.add_argument('--hostname', required=True,
148			help='Hostname or IP address')
149			parser_ssh.add_argument('--port', required=False,
150			help='SSH port (default: %(default)s)')
151			parser_ssh.add_argument('--ssh-username',
152			help='SSH username (default: %(default)r)')
153			parser_ssh.add_argument('--ssh-password',
154			help='SSH password (default: %(default)r)')
155
156			parser_ssh.set_defaults(
157			port=int(self._config.get('ssh', 'port', fallback=22)),
158			ssh_username=self._config.get('ssh', 'username', fallback='gmp'),
159			ssh_password=self._config.get('ssh', 'password', fallback='gmp'),
160			)
161
162			parser_tls = self._subparsers.add_parser(
163			'tls', help='Use TLS secured connection to connect to service',
164			parents=[self._root_parser])
165			parser_tls.add_argument('--hostname', required=True,
166			help='Hostname or IP address')
167			parser_tls.add_argument('--port', required=False,
168			help='GMP/OSP port (default: %(default)s)')
169			parser_tls.add_argument(
170			'--certfile', required=False,
171			help='Path to the certificate file for client authentication. '
172			'(default: %(default)s)')
173			parser_tls.add_argument(
174			'--keyfile', required=False,
175			help='Path to key file for client authentication. '
176			'(default: %(default)s)')
177			parser_tls.add_argument(
178			'--cafile', required=False,
179			help='Path to CA certificate for server authentication. '
180			'(default: %(default)s)')
181			parser_tls.add_argument(
182			'--no-credentials', required=False, default=False,
183			help='Use only certificates for authentication')
184			parser_tls.set_defaults(
185			port=int(self._config.get(
186			'tls', 'port', fallback=DEFAULT_GVM_PORT)),
187			certfile=self._config.get('tls', 'certfile', fallback=None),
188			keyfile=self._config.get('tls', 'keyfile', fallback=None),
189			cafile=self._config.get('tls', 'cafile', fallback=None),
190			)
191
192			parser_socket = self._subparsers.add_parser(
193			'socket', help='Use UNIX Domain socket to connect to service',
194			parents=[self._root_parser])
195
196			socketpath_group = parser_socket.add_mutually_exclusive_group()
197			socketpath_group.add_argument(
198			'--sockpath', nargs='?', default=None,
199			help='Deprecated, use --socketpath instead')
200			socketpath_group.add_argument(
201			'--socketpath', nargs='?',
202			help='Path to UNIX Domain socket (default: %(default)s)')
203
204			parser_socket.set_defaults(
205			socketpath=self._config.get(
206			'unixsocket', 'socketpath', fallback=DEFAULT_UNIX_SOCKET_PATH),
207			)
208
209			def add_protocol_argument(self):
210			self.add_argument(
211			'--protocol', required=False, default=DEFAULT_PROTOCOL,
212			choices=[PROTOCOL_GMP, PROTOCOL_OSP],
213			help='Service protocol to use (default: %(default)s)')
214
215
216			def create_parser(description, logfilename):
217			return CliParser(description, logfilename)
218
219
220			def create_connection(connection_type, socketpath=None, timeout=None,
221			hostname=None, port=None, certfile=None, keyfile=None,
222			cafile=None, ssh_username=None, ssh_password=None,
223			**kwargs):
224			if 'socket' in connection_type:
225			return UnixSocketConnection(
226			timeout=timeout,
227			path=socketpath
228			)
229
230			if 'tls' in connection_type:
231			return TLSConnection(
232			timeout=timeout,
233			hostname=hostname,
234			port=port,
235			certfile=certfile,
236			keyfile=keyfile,
237			cafile=cafile,
238			)
239
240			return SSHConnection(
241			timeout=timeout,
242			hostname=hostname,
243			port=port,
244			username=ssh_username,
245			password=ssh_password
246			)
247

greenbone / gvm-tools

Pull Request — master (#140)

gvmtools.parser.CliParser._add_subparsers() B

Complexity

Size

Duplication

Importance

How to fix Long Method

Long Method

Duplication Side-by-Side

Filter issues like