ospd.parser.CliParser.__init__() - Code Metrics - greenbone/ospd - Measure and Improve Code Quality continuously with Scrutinizer

ospd.parser.CliParser.init() B
last analyzed 2021-07-12 06:54 UTC

↳ Parent: ospd.parser

Complexity

Conditions

Size

Total Lines	143
Code Lines	109

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	109
nop	2
dl	0
loc	143
rs	7
c	0
b	0
f	0

How to fix Long Method

# Copyright (C) 2014-2021 Greenbone Networks GmbH
#
# SPDX-License-Identifier: AGPL-3.0-or-later
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero 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 Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.

import argparse
import logging
from pathlib import Path

from ospd.config import Config

# Default file locations as used by a OpenVAS default installation
DEFAULT_KEY_FILE = "/usr/var/lib/gvm/private/CA/serverkey.pem"
DEFAULT_CERT_FILE = "/usr/var/lib/gvm/CA/servercert.pem"
DEFAULT_CA_FILE = "/usr/var/lib/gvm/CA/cacert.pem"

DEFAULT_PORT = 0
DEFAULT_ADDRESS = "0.0.0.0"
DEFAULT_NICENESS = 10
DEFAULT_UNIX_SOCKET_MODE = "0o700"
DEFAULT_CONFIG_PATH = "~/.config/ospd.conf"
DEFAULT_LOG_CONFIG_PATH = "~/.config/ospd-logging.conf"
DEFAULT_UNIX_SOCKET_PATH = "/var/run/ospd/ospd.sock"
DEFAULT_PID_PATH = "/var/run/ospd.pid"
DEFAULT_LOCKFILE_DIR_PATH = "/var/run/ospd"
DEFAULT_STREAM_TIMEOUT = 10  # ten seconds
DEFAULT_SCANINFO_STORE_TIME = 0  # in hours
DEFAULT_MAX_SCAN = 0  # 0 = disable
DEFAULT_MIN_FREE_MEM_SCAN_QUEUE = 0  # 0 = Disable
DEFAULT_MAX_QUEUED_SCANS = 0  # 0 = Disable

ParserType = argparse.ArgumentParser
Arguments = argparse.Namespace

logger = logging.getLogger(__name__)


class CliParser:
    def __init__(self, description: str) -> None:
        """Create a command-line arguments parser for OSPD."""
        self._name = description
        parser = argparse.ArgumentParser(description=description)

        parser.add_argument(
            '--version', action='store_true', help='Print version then exit.'
        )

        parser.add_argument(
            '-s',
            '--config',
            nargs='?',
            default=DEFAULT_CONFIG_PATH,
            help='Configuration file path (default: %(default)s)',
        )
        parser.add_argument(
            '--log-config',
            nargs='?',
            default=DEFAULT_LOG_CONFIG_PATH,
            help='Log configuration file path (default: %(default)s)',
        )
        parser.add_argument(
            '-p',
            '--port',
            default=DEFAULT_PORT,
            type=self.network_port,
            help='TCP Port to listen on. Default: %(default)s',
        )
        parser.add_argument(
            '-b',
            '--bind-address',
            default=DEFAULT_ADDRESS,
            dest='address',
            help='Address to listen on. Default: %(default)s',
        )
        parser.add_argument(
            '-u',
            '--unix-socket',
            default=DEFAULT_UNIX_SOCKET_PATH,
            help='Unix file socket to listen on. Default: %(default)s',
        )
        parser.add_argument(
            '--pid-file',
            default=DEFAULT_PID_PATH,
            help='Location of the file for the process ID. '
            'Default: %(default)s',
        )
        parser.add_argument(
            '--lock-file-dir',
            default=DEFAULT_LOCKFILE_DIR_PATH,
            help='Directory where lock files are placed. Default: %(default)s',
        )
        parser.add_argument(
            '-m',
            '--socket-mode',
            default=DEFAULT_UNIX_SOCKET_MODE,
            help='Unix file socket mode. Default: %(default)s',
        )
        parser.add_argument(
            '-k',
            '--key-file',
            default=DEFAULT_KEY_FILE,
            help='Server key file. Default: %(default)s',
        )
        parser.add_argument(
            '-c',
            '--cert-file',
            default=DEFAULT_CERT_FILE,
            help='Server cert file. Default: %(default)s',
        )
        parser.add_argument(
            '--ca-file',
            default=DEFAULT_CA_FILE,
            help='CA cert file. Default: %(default)s',
        )
        parser.add_argument(
            '-L',
            '--log-level',
            default='INFO',
            type=self.log_level,
            help='Wished level of logging. Default: %(default)s',
        )
        parser.add_argument(
            '-f',
            '--foreground',
            action='store_true',
            help='Run in foreground and logs all messages to console.',
        )
        parser.add_argument(
            '-t',
            '--stream-timeout',
            default=DEFAULT_STREAM_TIMEOUT,
            type=int,
            help='Stream timeout. Default: %(default)s',
        )
        parser.add_argument(
            '-l', '--log-file', help='Path to the logging file.'
        )
        parser.add_argument(
            '--niceness',
            default=DEFAULT_NICENESS,
            type=int,
            help='Start the scan with the given niceness. Default %(default)s',
        )
        parser.add_argument(
            '--scaninfo-store-time',
            default=DEFAULT_SCANINFO_STORE_TIME,
            type=int,
            help='Time in hours a scan is stored before being considered '
            'forgotten and being delete from the scan table. '
            'Default %(default)s, disabled.',
        )
        parser.add_argument(
            '--list-commands',
            action='store_true',
            help='Display all protocol commands',
        )
        parser.add_argument(
            '--max-scans',
            default=DEFAULT_MAX_SCAN,
            type=int,
            help='Max. amount of parallel task that can be started. '
            'Default %(default)s, disabled',
        )
        parser.add_argument(
            '--min-free-mem-scan-queue',
            default=DEFAULT_MIN_FREE_MEM_SCAN_QUEUE,
            type=int,
            help='Minimum free memory in MB required to run the scan. '
            'If no enough free memory is available, the scan queued. '
            'Default %(default)s, disabled',
        )
        parser.add_argument(
            '--max-queued-scans',
            default=DEFAULT_MAX_QUEUED_SCANS,
            type=int,
            help='Maximum number allowed of queued scans before '
            'starting to reject new scans. '
            'Default %(default)s, disabled',
        )

        self.parser = parser

    def network_port(self, string: str) -> int:
        """Check if provided string is a valid network port."""

        value = int(string)
        if not 0 < value <= 65535:
            raise argparse.ArgumentTypeError(
                'port must be in ]0,65535] interval'
            )
        return value

    def log_level(self, string: str) -> str:
        """Check if provided string is a valid log level."""

        if not hasattr(logging, string.upper()):
            raise argparse.ArgumentTypeError(
                'log level must be one of {debug,info,warning,error,critical}'
            )
        return string.upper()

    def _set_defaults(self, configfilename=None) -> None:
        self._config = self._load_config(configfilename)
        self.parser.set_defaults(**self._config.defaults())

    def _load_config(self, configfile: str) -> Config:
        config = Config()

        if not configfile:
            return config

        configpath = Path(configfile)

        if not configpath.expanduser().resolve().exists():
            logger.debug('Ignoring non existing config file %s', configfile)
            return config

        try:
            config.load(configpath, def_section=self._name)
            logger.debug('Loaded config %s', configfile)
        except Exception as e:  # pylint: disable=broad-except
            raise RuntimeError(
                'Error while parsing config file {config}. Error was '
                '{message}'.format(config=configfile, message=e)
            ) from None

        return config

    def parse_arguments(self, args=None):
        # Parse args to get the config file path passed as option
        _args, _ = self.parser.parse_known_args(args)

        # Load the defaults from the config file if it exists.
        # This override also what it was passed as cmd option.
        self._set_defaults(_args.config)
        args, _ = self.parser.parse_known_args(args)

        return args


def create_parser(description: str) -> CliParser:
    return CliParser(description)


1			# Copyright (C) 2014-2021 Greenbone Networks GmbH
2			#
3			# SPDX-License-Identifier: AGPL-3.0-or-later
4			#
5			# This program is free software: you can redistribute it and/or modify
6			# it under the terms of the GNU Affero General Public License as
7			# published by the Free Software Foundation, either version 3 of the
8			# License, or (at your option) any later version.
9			#
10			# This program is distributed in the hope that it will be useful,
11			# but WITHOUT ANY WARRANTY; without even the implied warranty of
12			# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
13			# GNU Affero General Public License for more details.
14			#
15			# You should have received a copy of the GNU Affero General Public License
16			# along with this program. If not, see <http://www.gnu.org/licenses/>.
17
18			import argparse
19			import logging
20			from pathlib import Path
21
22			from ospd.config import Config
23
24			# Default file locations as used by a OpenVAS default installation
25			DEFAULT_KEY_FILE = "/usr/var/lib/gvm/private/CA/serverkey.pem"
26			DEFAULT_CERT_FILE = "/usr/var/lib/gvm/CA/servercert.pem"
27			DEFAULT_CA_FILE = "/usr/var/lib/gvm/CA/cacert.pem"
28
29			DEFAULT_PORT = 0
30			DEFAULT_ADDRESS = "0.0.0.0"
31			DEFAULT_NICENESS = 10
32			DEFAULT_UNIX_SOCKET_MODE = "0o700"
33			DEFAULT_CONFIG_PATH = "~/.config/ospd.conf"
34			DEFAULT_LOG_CONFIG_PATH = "~/.config/ospd-logging.conf"
35			DEFAULT_UNIX_SOCKET_PATH = "/var/run/ospd/ospd.sock"
36			DEFAULT_PID_PATH = "/var/run/ospd.pid"
37			DEFAULT_LOCKFILE_DIR_PATH = "/var/run/ospd"
38			DEFAULT_STREAM_TIMEOUT = 10 # ten seconds
39			DEFAULT_SCANINFO_STORE_TIME = 0 # in hours
40			DEFAULT_MAX_SCAN = 0 # 0 = disable
41			DEFAULT_MIN_FREE_MEM_SCAN_QUEUE = 0 # 0 = Disable
42			DEFAULT_MAX_QUEUED_SCANS = 0 # 0 = Disable
43
44			ParserType = argparse.ArgumentParser
45			Arguments = argparse.Namespace
46
47			logger = logging.getLogger(__name__)
48
49
50			class CliParser:
51			def __init__(self, description: str) -> None:
52			"""Create a command-line arguments parser for OSPD."""
53			self._name = description
54			parser = argparse.ArgumentParser(description=description)
55
56			parser.add_argument(
57			'--version', action='store_true', help='Print version then exit.'
58			)
59
60			parser.add_argument(
61			'-s',
62			'--config',
63			nargs='?',
64			default=DEFAULT_CONFIG_PATH,
65			help='Configuration file path (default: %(default)s)',
66			)
67			parser.add_argument(
68			'--log-config',
69			nargs='?',
70			default=DEFAULT_LOG_CONFIG_PATH,
71			help='Log configuration file path (default: %(default)s)',
72			)
73			parser.add_argument(
74			'-p',
75			'--port',
76			default=DEFAULT_PORT,
77			type=self.network_port,
78			help='TCP Port to listen on. Default: %(default)s',
79			)
80			parser.add_argument(
81			'-b',
82			'--bind-address',
83			default=DEFAULT_ADDRESS,
84			dest='address',
85			help='Address to listen on. Default: %(default)s',
86			)
87			parser.add_argument(
88			'-u',
89			'--unix-socket',
90			default=DEFAULT_UNIX_SOCKET_PATH,
91			help='Unix file socket to listen on. Default: %(default)s',
92			)
93			parser.add_argument(
94			'--pid-file',
95			default=DEFAULT_PID_PATH,
96			help='Location of the file for the process ID. '
97			'Default: %(default)s',
98			)
99			parser.add_argument(
100			'--lock-file-dir',
101			default=DEFAULT_LOCKFILE_DIR_PATH,
102			help='Directory where lock files are placed. Default: %(default)s',
103			)
104			parser.add_argument(
105			'-m',
106			'--socket-mode',
107			default=DEFAULT_UNIX_SOCKET_MODE,
108			help='Unix file socket mode. Default: %(default)s',
109			)
110			parser.add_argument(
111			'-k',
112			'--key-file',
113			default=DEFAULT_KEY_FILE,
114			help='Server key file. Default: %(default)s',
115			)
116			parser.add_argument(
117			'-c',
118			'--cert-file',
119			default=DEFAULT_CERT_FILE,
120			help='Server cert file. Default: %(default)s',
121			)
122			parser.add_argument(
123			'--ca-file',
124			default=DEFAULT_CA_FILE,
125			help='CA cert file. Default: %(default)s',
126			)
127			parser.add_argument(
128			'-L',
129			'--log-level',
130			default='INFO',
131			type=self.log_level,
132			help='Wished level of logging. Default: %(default)s',
133			)
134			parser.add_argument(
135			'-f',
136			'--foreground',
137			action='store_true',
138			help='Run in foreground and logs all messages to console.',
139			)
140			parser.add_argument(
141			'-t',
142			'--stream-timeout',
143			default=DEFAULT_STREAM_TIMEOUT,
144			type=int,
145			help='Stream timeout. Default: %(default)s',
146			)
147			parser.add_argument(
148			'-l', '--log-file', help='Path to the logging file.'
149			)
150			parser.add_argument(
151			'--niceness',
152			default=DEFAULT_NICENESS,
153			type=int,
154			help='Start the scan with the given niceness. Default %(default)s',
155			)
156			parser.add_argument(
157			'--scaninfo-store-time',
158			default=DEFAULT_SCANINFO_STORE_TIME,
159			type=int,
160			help='Time in hours a scan is stored before being considered '
161			'forgotten and being delete from the scan table. '
162			'Default %(default)s, disabled.',
163			)
164			parser.add_argument(
165			'--list-commands',
166			action='store_true',
167			help='Display all protocol commands',
168			)
169			parser.add_argument(
170			'--max-scans',
171			default=DEFAULT_MAX_SCAN,
172			type=int,
173			help='Max. amount of parallel task that can be started. '
174			'Default %(default)s, disabled',
175			)
176			parser.add_argument(
177			'--min-free-mem-scan-queue',
178			default=DEFAULT_MIN_FREE_MEM_SCAN_QUEUE,
179			type=int,
180			help='Minimum free memory in MB required to run the scan. '
181			'If no enough free memory is available, the scan queued. '
182			'Default %(default)s, disabled',
183			)
184			parser.add_argument(
185			'--max-queued-scans',
186			default=DEFAULT_MAX_QUEUED_SCANS,
187			type=int,
188			help='Maximum number allowed of queued scans before '
189			'starting to reject new scans. '
190			'Default %(default)s, disabled',
191			)
192
193			self.parser = parser
194
195			def network_port(self, string: str) -> int:
196			"""Check if provided string is a valid network port."""
197
198			value = int(string)
199			if not 0 < value <= 65535:
200			raise argparse.ArgumentTypeError(
201			'port must be in ]0,65535] interval'
202			)
203			return value
204
205			def log_level(self, string: str) -> str:
206			"""Check if provided string is a valid log level."""
207
208			if not hasattr(logging, string.upper()):
209			raise argparse.ArgumentTypeError(
210			'log level must be one of {debug,info,warning,error,critical}'
211			)
212			return string.upper()
213
214			def _set_defaults(self, configfilename=None) -> None:
215			self._config = self._load_config(configfilename)
216			self.parser.set_defaults(**self._config.defaults())
217
218			def _load_config(self, configfile: str) -> Config:
219			config = Config()
220
221			if not configfile:
222			return config
223
224			configpath = Path(configfile)
225
226			if not configpath.expanduser().resolve().exists():
227			logger.debug('Ignoring non existing config file %s', configfile)
228			return config
229
230			try:
231			config.load(configpath, def_section=self._name)
232			logger.debug('Loaded config %s', configfile)
233			except Exception as e: # pylint: disable=broad-except
234			raise RuntimeError(
235			'Error while parsing config file {config}. Error was '
236			'{message}'.format(config=configfile, message=e)
237			) from None
238
239			return config
240
241			def parse_arguments(self, args=None):
242			# Parse args to get the config file path passed as option
243			_args, _ = self.parser.parse_known_args(args)
244
245			# Load the defaults from the config file if it exists.
246			# This override also what it was passed as cmd option.
247			self._set_defaults(_args.config)
248			args, _ = self.parser.parse_known_args(args)
249
250			return args
251
252
253			def create_parser(description: str) -> CliParser:
254			return CliParser(description)
255

greenbone / ospd

ospd.parser.CliParser.__init__() B last analyzed 2021-07-12 06:54 UTC

Complexity

Size

Duplication

Importance

How to fix Long Method

Long Method

Duplication Side-by-Side

Filter issues like

ospd.parser.CliParser.init() B
last analyzed 2021-07-12 06:54 UTC