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

ospd.main.main() C
last analyzed 2021-07-12 06:54 UTC

↳ Parent: ospd.main

Complexity

Conditions

Size

Total Lines	76
Code Lines	52

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	9
eloc	52
nop	3
dl	0
loc	76
rs	6.2375
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 logging

import os
import sys
import atexit
import signal

from functools import partial

from typing import Type, Optional
from pathlib import Path

from ospd.misc import go_to_background, create_pid
from ospd.ospd import OSPDaemon
from ospd.parser import create_parser, ParserType
from ospd.server import TlsServer, UnixSocketServer, BaseServer
from ospd.logger import init_logging


COPYRIGHT = """Copyright (C) 2014-2021 Greenbone Networks GmbH
License GPLv2+: GNU GPL version 2 or later
This is free software: you are free to change and redistribute it.
There is NO WARRANTY, to the extent permitted by law."""

LOGGER = logging.getLogger(__name__)


def print_version(daemon: OSPDaemon, file=sys.stdout):
    """Prints the server version and license information."""

    scanner_name = daemon.get_scanner_name()
    server_version = daemon.get_server_version()
    protocol_version = daemon.get_protocol_version()
    daemon_name = daemon.get_daemon_name()
    daemon_version = daemon.get_daemon_version()

    print(
        "OSP Server for {0}: {1}".format(scanner_name, server_version),
        file=file,
    )
    print("OSP: {0}".format(protocol_version), file=file)
    print("{0}: {1}".format(daemon_name, daemon_version), file=file)
    print(file=file)
    print(COPYRIGHT, file=file)


def exit_cleanup(
    pidfile: str,
    server: BaseServer,
    daemon: OSPDaemon,
    _signum=None,
    _frame=None,
) -> None:
    """Removes the pidfile before ending the daemon."""
    signal.signal(signal.SIGINT, signal.SIG_IGN)
    pidpath = Path(pidfile)

    if not pidpath.is_file():
        return

    with pidpath.open() as f:
        if int(f.read()) == os.getpid():
            LOGGER.debug("Performing exit clean up")
            daemon.daemon_exit_cleanup()
            LOGGER.info("Shutting-down server ...")
            server.close()
            LOGGER.debug("Finishing daemon process")
            pidpath.unlink()
            sys.exit()


def main(
    name: str,
    daemon_class: Type[OSPDaemon],
    parser: Optional[ParserType] = None,
):
    """OSPD Main function."""

    if not parser:
        parser = create_parser(name)
    args = parser.parse_arguments()

    if args.version:
        args.foreground = True

    init_logging(
        args.log_level,
        log_file=args.log_file,
        log_config=args.log_config,
        foreground=args.foreground,
    )

    if args.port == 0:
        server = UnixSocketServer(
            args.unix_socket,
            args.socket_mode,
            args.stream_timeout,
        )
    else:
        server = TlsServer(
            args.address,
            args.port,
            args.cert_file,
            args.key_file,
            args.ca_file,
            args.stream_timeout,
        )

    daemon = daemon_class(**vars(args))

    if args.version:
        print_version(daemon)
        sys.exit()

    if args.list_commands:
        print(daemon.get_help_text())
        sys.exit()

    if not args.foreground:
        go_to_background()

    if not create_pid(args.pid_file):
        sys.exit()

    # Set signal handler and cleanup
    atexit.register(
        exit_cleanup, pidfile=args.pid_file, server=server, daemon=daemon
    )
    signal.signal(
        signal.SIGTERM, partial(exit_cleanup, args.pid_file, server, daemon)
    )
    signal.signal(
        signal.SIGINT, partial(exit_cleanup, args.pid_file, server, daemon)
    )
    if not daemon.check():
        return 1

    LOGGER.info(
        "Starting %s version %s.",
        daemon.daemon_info['name'],
        daemon.daemon_info['version'],
    )

    daemon.init(server)
    daemon.run()

    return 0


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 logging
19
20			import os
21			import sys
22			import atexit
23			import signal
24
25			from functools import partial
26
27			from typing import Type, Optional
28			from pathlib import Path
29
30			from ospd.misc import go_to_background, create_pid
31			from ospd.ospd import OSPDaemon
32			from ospd.parser import create_parser, ParserType
33			from ospd.server import TlsServer, UnixSocketServer, BaseServer
34			from ospd.logger import init_logging
35
36
37			COPYRIGHT = """Copyright (C) 2014-2021 Greenbone Networks GmbH
38			License GPLv2+: GNU GPL version 2 or later
39			This is free software: you are free to change and redistribute it.
40			There is NO WARRANTY, to the extent permitted by law."""
41
42			LOGGER = logging.getLogger(__name__)
43
44
45			def print_version(daemon: OSPDaemon, file=sys.stdout):
46			"""Prints the server version and license information."""
47
48			scanner_name = daemon.get_scanner_name()
49			server_version = daemon.get_server_version()
50			protocol_version = daemon.get_protocol_version()
51			daemon_name = daemon.get_daemon_name()
52			daemon_version = daemon.get_daemon_version()
53
54			print(
55			"OSP Server for {0}: {1}".format(scanner_name, server_version),
56			file=file,
57			)
58			print("OSP: {0}".format(protocol_version), file=file)
59			print("{0}: {1}".format(daemon_name, daemon_version), file=file)
60			print(file=file)
61			print(COPYRIGHT, file=file)
62
63
64			def exit_cleanup(
65			pidfile: str,
66			server: BaseServer,
67			daemon: OSPDaemon,
68			_signum=None,
69			_frame=None,
70			) -> None:
71			"""Removes the pidfile before ending the daemon."""
72			signal.signal(signal.SIGINT, signal.SIG_IGN)
73			pidpath = Path(pidfile)
74
75			if not pidpath.is_file():
76			return
77
78			with pidpath.open() as f:
79			if int(f.read()) == os.getpid():
80			LOGGER.debug("Performing exit clean up")
81			daemon.daemon_exit_cleanup()
82			LOGGER.info("Shutting-down server ...")
83			server.close()
84			LOGGER.debug("Finishing daemon process")
85			pidpath.unlink()
86			sys.exit()
87
88
89			def main(
90			name: str,
91			daemon_class: Type[OSPDaemon],
92			parser: Optional[ParserType] = None,
93			):
94			"""OSPD Main function."""
95
96			if not parser:
97			parser = create_parser(name)
98			args = parser.parse_arguments()
99
100			if args.version:
101			args.foreground = True
102
103			init_logging(
104			args.log_level,
105			log_file=args.log_file,
106			log_config=args.log_config,
107			foreground=args.foreground,
108			)
109
110			if args.port == 0:
111			server = UnixSocketServer(
112			args.unix_socket,
113			args.socket_mode,
114			args.stream_timeout,
115			)
116			else:
117			server = TlsServer(
118			args.address,
119			args.port,
120			args.cert_file,
121			args.key_file,
122			args.ca_file,
123			args.stream_timeout,
124			)
125
126			daemon = daemon_class(**vars(args))
127
128			if args.version:
129			print_version(daemon)
130			sys.exit()
131
132			if args.list_commands:
133			print(daemon.get_help_text())
134			sys.exit()
135
136			if not args.foreground:
137			go_to_background()
138
139			if not create_pid(args.pid_file):
140			sys.exit()
141
142			# Set signal handler and cleanup
143			atexit.register(
144			exit_cleanup, pidfile=args.pid_file, server=server, daemon=daemon
145			)
146			signal.signal(
147			signal.SIGTERM, partial(exit_cleanup, args.pid_file, server, daemon)
148			)
149			signal.signal(
150			signal.SIGINT, partial(exit_cleanup, args.pid_file, server, daemon)
151			)
152			if not daemon.check():
153			return 1
154
155			LOGGER.info(
156			"Starting %s version %s.",
157			daemon.daemon_info['name'],
158			daemon.daemon_info['version'],
159			)
160
161			daemon.init(server)
162			daemon.run()
163
164			return 0
165

greenbone / ospd

ospd.main.main() C 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.main.main() C
last analyzed 2021-07-12 06:54 UTC