From 1a5e60fc1069d39db1b0935f4ac8e4cace83fb76 Mon Sep 17 00:00:00 2001 From: Simeon Simeonov Date: Fri, 10 Apr 2020 23:27:15 +0200 Subject: Remove Python2 support and cleanup the code and the documentation --- INSTALL.md | 34 ++-- README.md | 42 +++-- beinc_generic_client.py | 273 +++++++++++++------------------ beinc_poller.py | 238 --------------------------- beinc_pull.py | 248 +++++++++++++++++++++++++++++ beinc_server.py | 255 ++++++++++++++--------------- beinc_weechat.py | 416 +++++++++++++++++++++++------------------------- 7 files changed, 724 insertions(+), 782 deletions(-) delete mode 100755 beinc_poller.py create mode 100755 beinc_pull.py diff --git a/INSTALL.md b/INSTALL.md index b0e8a0a..15bcb35 100644 --- a/INSTALL.md +++ b/INSTALL.md @@ -1,38 +1,38 @@ -## Installation Instructions +# Installation Instructions Refer to README for a basic information about the different BEINC components and their software requirements! -### Installing the dependencies +## Installing the dependencies -#### Gentoo GNU/Linux 2014, 2015 +### Gentoo GNU/Linux -# emerge -va dev-python/pyopenssl dev-python/notify-python + ```bash + emerge -va dev-python/notify2 + ``` -#### Ubuntu GNU/Linux +### Ubuntu GNU/Linux -# apt-get install python-openssl python-notify2 + ```bash + apt-get install python-notify2 + ``` -#### FreeBSD 10, 11 - -# cd /usr/ports/security/py-openssl -# make install clean +### FreeBSD 12 # cd /usr/ports/devel/py-notify # make install clean -#### Intallation using virtualenv and pip +### Intallation using virtualenv and pip Very often you are not the administrator of the server hosting your BEINC server This is where virtualenv may become handy (provided that it has been installed): $ virtualenv beinc_installation $ cd beinc_installation $ source bin/activate -$ pip install pyopenssl $ cd beinc- $ ./beinc_server.py -h @@ -59,13 +59,13 @@ The example loads the server listening on 10.0.0.44, port 9678 using the default ~/.beinc_server.json as a configuration file and usually gives you the Twisted log console -beinc_poller.py: -$ ./beinc_poller.py -h +beinc_pull.py: +$ ./beinc_pull.py -h The example lists all available command options -$ ./beinc_poller.py -n weechat -t 10 -f 7 --cert-file /home/sgs/weechat_cert.crt \ +$ ./beinc_pull.py -n weechat -t 10 -f 7 --cert-file /home/sgs/weechat_cert.crt \ -p /home/sgs/.weechat_beinc_passwd https://10.0.0.44:9678 -Starts polling from the instance (target) defined above, -defining an OSD timeout of 10 seconds, polling frequency of 7 seconds, +Starts pulling from the instance (target) defined above, +defining an OSD timeout of 10 seconds, pulling frequency of 7 seconds, checking whether the server's SSL certificate was signed (issued) by /home/sgs/weechat_cert.crt, extracting the target's password from /home/sgs/.weechat_beinc_passwd diff --git a/README.md b/README.md index 388562b..200e29d 100644 --- a/README.md +++ b/README.md @@ -1,4 +1,4 @@ -Copyright (C) 2014-2018 - Simeon Simeonov +Copyright (C) 2014-2020 - Simeon Simeonov See the end of the file for license conditions. @@ -10,14 +10,14 @@ on-screen-display notification scenarios. The current version of BEINC contains the following components: beinc_server.py - server used for queueing or providing on-screen-display (OSD) -beinc_poller.py - client used to fetch enqueued messages from beinc_server.py - and provide OSD +beinc_pull.py - client used to fetch enqueued messages from beinc_server.py + and provide OSD beinc_weechat.py - a complete script / client for the Weechat IRC client >=0.4.0 used to push notification messages to beinc_server.py beinc_generic_client.py - a simple client used to push notification messages to beinc_server.py Its main purpose is to provide a convenient way to - test beinc_server.py (and beinc_poller.py) as well as + test beinc_server.py (and beinc_pull.py) as well as to serve as an example for how to develop BEINC-clients + documentation and a sample configuration file (beinc_config_sample.json) @@ -28,7 +28,7 @@ BEINC is a free-software, licensed under the GPL3. It gives you the freedoms of using it, studying it and modifying it. BEINC is designed to assist you two common scenarios. -As an example we will consider the case of an IRC cliet, although BEINC can be +As an example we will consider the case of an IRC client, although BEINC can be used by any client that conforms with BEINC's "messaging protocol". We have the common situation where a user is accessing an IRC client located on @@ -66,20 +66,20 @@ Solution: - set up the beinc_server.py on it. No OSD capabilities are needed (no X11). Define instance(s) for queueing. Define a queue size (how many notifications to store) for each of them. - - set up beinc_poller.py on the desktop workstation (the computer where the + - set up beinc_pull.py on the desktop workstation (the computer where the notification-message should be displayed). - beinc_poller.py should be able to access the server running beinc_server.py - Polling interval of 5 seconds should be enough for most users. + beinc_pull.py should be able to access the server running beinc_server.py + Pulling interval of 5 seconds should be enough for most users. Read beinc_config_sample.json.readme for details on how to setup beinc_server.py and beinc_weechat.py! Use "-h" command line parameter to display the available options for -beinc_poller.py and beinc_generic_client.py +beinc_pull.py and beinc_generic_client.py Each configured beinc_server.py instance has an unique name. There are 2 operations that can be done on an instance: - push - a client is sending a notification to the BEINC server. A push is available for all instances. - - pull - a poller (f.i. beinc_poller.py) is fetching data from the instance-queue. + - pull - a pull (f.i. beinc_pull.py) is fetching data from the instance-queue. A pull is available only for instances defined for queueing. (read beinc_config_sample.json.readme for details!) Example: @@ -91,28 +91,38 @@ your URL will be: https://hostname:port Any system running the software required for the selected components. All components tested on: Gentoo GNU/Linux, - Ubuntu GNU/Linux 16.4, 18.4, - FreeBSD 10.x, 11.x + Ubuntu GNU/Linux 18.4, + FreeBSD 12.x ### Requirements -All components: Python >= 2.7.9 or Python >= 3.4.* +All components: Python >= 3.6.* beinc_server.py: pynotify >= 0.1 (optional) beinc_weechat.py: Weechat >= 0.4.0 -beinc_poller.py: pynotify >= 0.1 +beinc_pull.py: pynotify >= 0.1 beinc_generic_client.py: No additional software required Read INSTALL in this very same folder for more details about installing the requirements! -### License +## Support and contributing + +BEINC is hosted on GitHub: https://github.com/blackm0re/BEINC + + +## Author + +Simeon Simeonov - sgs @ Freenode + + +## License This file is part of BEINC. BEINC 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 +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, diff --git a/beinc_generic_client.py b/beinc_generic_client.py index 4d8fc35..5514642 100755 --- a/beinc_generic_client.py +++ b/beinc_generic_client.py @@ -1,8 +1,8 @@ #!/usr/bin/env python # -*- coding: utf-8 -*- -# Blackmore's Enhanced IRC-Notification Collection (BEINC) v2.0 -# Copyright (C) 2013-2018 Simeon Simeonov +# Blackmore's Enhanced IRC-Notification Collection (BEINC) v4.0 +# Copyright (C) 2013-2020 Simeon Simeonov # 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 @@ -16,146 +16,114 @@ # You should have received a copy of the GNU General Public License # along with this program. If not, see . - - +"""A generic BEINC client that can be used for testing or as a template""" import argparse import errno import getpass -import httplib # for Python < 2.7.9 +import io +import json import os import socket import ssl import sys -import xmlrpclib +import urllib.parse +import urllib.request __author__ = 'Simeon Simeonov' -__version__ = '3.0' +__version__ = '4.0' __license__ = 'GPL3' -BEINC_SSL_METHODS = {'TLSv1': ssl.PROTOCOL_TLSv1} -try: - BEINC_SSL_METHODS.update({'TLSv1_1': ssl.PROTOCOL_TLSv1_1}) - BEINC_SSL_METHODS.update({'TLSv1_2': ssl.PROTOCOL_TLSv1_2}) -except: - pass - +def eprint(*arg, **kwargs): + """stdderr print wrapper""" + print(*arg, file=sys.stderr, flush=True, **kwargs) -class BEINCCustomHTTPSConnection(httplib.HTTPConnection): - """ - This class allows communication via SSL. - It is a reimplementation of httplib.HTTPSConnection and - allows the server certificate to be validated against CA - This functionality lacks in Python < 2.7.9 +def fetch_password(args_password): """ - default_port = httplib.HTTPS_PORT + Fetches the password from the provided `args_password` - def __init__(self, host, port=None, key_file=None, cert_file=None, - strict=None, timeout=socket._GLOBAL_DEFAULT_TIMEOUT, - source_address=None, custom_ssl_options={}): - httplib.HTTPConnection.__init__(self, host, port, strict, timeout, - source_address) - self.key_file = key_file - self.cert_file = cert_file - self.custom_ssl_options = custom_ssl_options + :param args_password: The password coming from argparse + :type args_password: str - def connect(self): - "Connect to a host on a given (SSL) port." - sock = socket.create_connection((self.host, self.port), - self.timeout, self.source_address) - if self._tunnel_host: - self.sock = sock - self._tunnel() - self.sock = ssl.wrap_socket(sock, - self.key_file, - self.cert_file, - **self.custom_ssl_options) + :return: The password string + :rtype: str + """ + if not args_password: + try: + return getpass.getpass() + except KeyboardInterrupt: + eprint(os.linesep + 'Prompt terminated') + sys.exit(errno.EACCES) + elif os.path.isfile(args_password): + try: + with io.open(args_password, 'r') as fp: + passwd = fp.readline() + if passwd.strip(): + return passwd.strip() + except Exception as e: + eprint(f'Unable to open password file: {e}') + sys.exit(1) + return args_password -class BEINCCustomSafeTransport(xmlrpclib.Transport): +def pull_notifications(ssl_context, args): """ - Provides an alternative HTTPS transport for clients running on Python<2.7.9 - """ - - def __init__(self, use_datetime=0, custom_ssl_options={}): - xmlrpclib.Transport.__init__(self, use_datetime=use_datetime) - self.custom_ssl_options = custom_ssl_options + Pulls the notifications from the BEINC server. - def make_connection(self, host): - if self._connection and host == self._connection[0]: - return self._connection[1] - try: - HTTPS = BEINCCustomHTTPSConnection - except AttributeError: - raise NotImplementedError( - "your version of httplib doesn't support HTTPS" - ) - else: - chost, self._extra_headers, x509 = self.get_host_info(host) - self._connection = host, HTTPS( - chost, - None, - custom_ssl_options=self.custom_ssl_options, - **(x509 or {})) - return self._connection[1] + :param ssl_context: The SSL context object + :type ssl_context: ssl.SSLContext + :param args: The arguments assigned from argparse + :type args: argparse.Namespace -def action_execute(args): + :return: The notifications (if any) + :rtype: list """ + response = urllib.request.urlopen( + args.url, + data=urllib.parse.urlencode( + ( + ('resource_name', args.rname), + ('password', args.password) + )).encode('utf-8'), + timeout=args.socket_timeout, + context=ssl_context) + response_dict = json.loads(response.read().decode('utf-8')) + if response.code != 200: + raise socket.error(response_dict.get('message', '')) + return response_dict['data']['messages'] + + +def push_notification(ssl_context, args): """ - try: - ssl_version = BEINC_SSL_METHODS.get(args.ssl_version, - ssl.PROTOCOL_SSLv23) - if sys.hexversion >= 0x20709f0: - # Python >= 2.7.9 - context = ssl.SSLContext(ssl_version) - context.verify_mode = ssl.CERT_NONE - if args.cert: - context.verify_mode = ssl.CERT_REQUIRED - context.load_verify_locations(os.path.expanduser(args.cert)) - context.check_hostname = bool(not args.disable_hostname_check) - if args.ciphers: - context.set_ciphers(args.ciphers) - transport = xmlrpclib.SafeTransport(context=context) - else: - # Python < 2.7.9 - ssl_options = {} - ssl_options['ssl_version'] = ssl_version - if args.cert: - ssl_options['ca_certs'] = os.path.expanduser(args.cert) - ssl_options['cert_reqs'] = ssl.CERT_REQUIRED - if args.ciphers: - ssl_options['ciphers'] = args.ciphers - transport = BEINCCustomSafeTransport( - custom_ssl_options=ssl_options) - server = xmlrpclib.ServerProxy(args.url, - transport=transport) - if args.pull: - print(server.pull(args.rname, args.password)) - else: - print(server.push(args.rname, - args.password, - args.title, - args.message)) - except xmlrpclib.Fault as fault: - sys.stderr.write( - 'BEINC server answered with errorCode={0}: {1}\n'.format( - fault.faultCode, - fault.faultString)) - except ssl.SSLError as e: - sys.stderr.write('BEINC SSL/TLS error: {0}\n'.format(e)) - sys.exit(errno.EPERM) - except socket.error as e: - sys.stderr.write('BEINC connection error: {0}\n'.format(e)) - sys.exit(errno.EPERM) - except Exception as e: - sys.stderr.write('BEINC generic client error: {0}\n'.format(e)) - sys.exit(errno.EPERM) + Pushes a single notification to the BEINC server. + :param ssl_context: The SSL context object + :type ssl_context: ssl.SSLContext -def main(): + :param args: The arguments assigned from argparse + :type args: argparse.Namespace + """ + response = urllib.request.urlopen( + args.url, + data=urllib.parse.urlencode( + ( + ('resource_name', args.rname), + ('password', args.password), + ('title', args.title), + ('message', args.message) + )).encode('utf-8'), + timeout=args.socket_timeout, + context=ssl_context) + response_dict = json.loads(response.read().decode('utf-8')) + if response.code != 200: + raise socket.error(response_dict.get('message', '')) + + +def main(inargs=None): + """main entry""" parser = argparse.ArgumentParser( description='The following options are available') parser.add_argument( @@ -178,13 +146,12 @@ def main(): dest='ciphers', default='', help='Preferred ciphers list (default: auto)') - if sys.hexversion >= 0x20709f0: - parser.add_argument( - '--disable-hostname-check', - action='store_true', - dest='disable_hostname_check', - default=False, - help='Do not check whether server cert matches server hostname') + parser.add_argument( + '--disable-hostname-check', + action='store_true', + dest='disable_hostname_check', + default=False, + help='Do not check whether server cert matches server hostname') parser.add_argument( '-m', '--message', metavar='MESSAGE', @@ -213,23 +180,6 @@ def main(): dest='pull', default=False, help='Perform a pull operation (default: push)') - if sys.hexversion >= 0x20709f0: - parser.add_argument( - '-s', '--ssl-version', - metavar='VERSION', - type=str, - dest='ssl_version', - default='auto', - help='Use SSL version: "auto" (default), ' - '"SSLv3", "TLSv1", "TLSv1_1", "TLSv1_2"') - else: - parser.add_argument( - '-s', '--ssl-version', - metavar='VERSION', - type=str, - dest='ssl_version', - default='auto', - help='Use SSL version: "auto" (default), "SSLv3", "TLSv1"') parser.add_argument( '-T', '--socket-timeout', metavar='SECONDS', @@ -247,29 +197,36 @@ def main(): parser.add_argument( '-v', '--version', action='version', - version='%(prog)s {0}'.format(__version__), + version=f'%(prog)s {__version__}', help='display program-version and exit') - args = parser.parse_args() - - if not args.password: - try: - args.password = getpass.getpass() - except Exception as e: - sys.stderr.write('Prompt terminated\n') - sys.exit(errno.EACCES) - elif os.path.isfile(args.password): - try: - with open(args.password, 'r') as fp: - passwd = fp.readline() - if passwd.strip(): - args.password = passwd.strip() - except Exception as e: - sys.stderr.write('Unable to open password file: {0}'.format(e)) - sys.exit(1) + args = parser.parse_args(inargs) if args.socket_timeout: socket.setdefaulttimeout(args.socket_timeout) - action_execute(args) - sys.exit(0) + args.password = fetch_password(args.password) + try: + context = ssl.SSLContext(ssl.PROTOCOL_TLSv1_2) + context.verify_mode = ssl.CERT_NONE + if args.cert: + context.verify_mode = ssl.CERT_REQUIRED + context.load_verify_locations(cafile=os.path.expanduser(args.cert)) + context.check_hostname = bool(not args.disable_hostname_check) + if args.ciphers: + context.set_ciphers(args.ciphers) + if args.pull: + print(json.dumps(pull_notifications(context, args), indent=4)) + else: + push_notification(context, args) + print('OK') + sys.exit(0) + except ssl.SSLError as e: + eprint(f'BEINC SSL/TLS error: {e}') + sys.exit(errno.EPERM) + except socket.error as e: + eprint(f'BEINC connection error: {e}') + sys.exit(errno.EPERM) + except Exception as e: + eprint(f'BEINC generic client error: {e}') + sys.exit(errno.EPERM) if __name__ == '__main__': diff --git a/beinc_poller.py b/beinc_poller.py deleted file mode 100755 index 624cd24..0000000 --- a/beinc_poller.py +++ /dev/null @@ -1,238 +0,0 @@ -#!/usr/bin/env python -# -*- coding: utf-8 -*- - -# Blackmore's Enhanced IRC-Notification Collection (BEINC) v3.0 -# Copyright (C) 2013-2018 Simeon Simeonov - -# 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 . - - -import argparse -import errno -import getpass -import json -import os -import sched -import socket -import ssl -import sys -import time - -PY2 = sys.version_info[0] == 2 -PY3 = sys.version_info[0] == 3 - -if PY3: - from urllib.parse import urlencode - from urllib.request import urlopen -else: - from urllib import urlencode - from urllib2 import urlopen - -try: - if PY3: - import notify2 as pynotify - else: - import pynotify -except ImportError as e: - pynotify = None - - -__author__ = 'Simeon Simeonov' -__version__ = '3.0' -__license__ = 'GPL3' - - -def display_notification(args, title, message): - """ - A wrapper function for displaying a single notification - - args: the argparse processed command-line arguments - title: notification title - message: notification message - """ - if args.osd_sys == 'pynotify': - if not pynotify: - raise Exception( - 'Could not load "pynotify".\n' - 'Please install "pynotify" or use a different osd-system!\n' - 'Terminating...\n') - if not pynotify.init('BEINC Notify'): - raise Exception('There was a problem with libnotify\n') - notification_obj = pynotify.Notification(summary=title, - message=message) - if PY3: - notification_obj.timeout = 1000 * args.osd_timeout - notification_obj.set_category('im.received') - else: - notification_obj.set_timeout(1000 * args.osd_timeout) - notification_obj.set_property( - 'app_name', - '{0} {1}'.format(sys.argv[0], __version__)) - notification_obj.show() - else: - raise Exception( - 'Unsupported osd-system: {0}\n'.format(args.osd_sys)) - - -def poll_notifications(scheduler, args): - """ - the core function initiated by the scheduler performing a single poll - - scheduler: scheduler object - args: the argparse processed command-line arguments - """ - try: - context = ssl.SSLContext(ssl.PROTOCOL_TLSv1_2) - context.verify_mode = ssl.CERT_NONE - if args.cert: - context.verify_mode = ssl.CERT_REQUIRED - context.load_verify_locations(cafile=os.path.expanduser(args.cert)) - context.check_hostname = bool(not args.disable_hostname_check) - if args.ciphers: - context.set_ciphers(args.ciphers) - response = urlopen( - args.url, - data=urlencode( - ( - ('resource_name', args.rname), - ('password', args.password) - )).encode('utf-8'), - timeout=args.socket_timeout, - context=context) - response_dict = json.loads(response.read().decode('utf-8')) - if response.code != 200: - raise socket.error(response_dict.get('message', '')) - for entry in response_dict['data']['messages']: - display_notification(args, - entry.get('title', ''), - entry.get('message', '')) - response.close() - scheduler.enter(args.frequency, - 1, - poll_notifications, - (scheduler, args)) - except ssl.SSLError as e: - sys.stderr.write('BEINC SSL/TLS error: {0}\n'.format(e)) - sys.exit(errno.EPERM) - except socket.error as e: - sys.stderr.write('BEINC connection error: {0}\n'.format(e)) - sys.exit(errno.EPERM) - except Exception as e: - sys.stderr.write('BEINC generic client error: {0}\n'.format(e)) - sys.exit(errno.EPERM) - - -def main(): - parser = argparse.ArgumentParser( - description='The following options are available') - parser.add_argument( - 'url', - metavar='URL', - type=str, - help='BEINC server destination URL') - parser.add_argument( - '-c', '--cert-file', - metavar='FILE', - type=str, - dest='cert', - default='', - help='CA-cert to check the server-cert against ' - '(default: Check disabled)') - parser.add_argument( - '--ciphers', - metavar='CIPHERS', - type=str, - dest='ciphers', - default='', - help='Preferred ciphers list (default: auto)') - parser.add_argument( - '--disable-hostname-check', - action='store_true', - dest='disable_hostname_check', - default=False, - help='Do not check whether server cert matches server hostname') - parser.add_argument( - '-f', '--frequency', - metavar='SECONDS', - type=int, - dest='frequency', - default=10, - help='Polling frequency in seconds (default: 10)') - parser.add_argument( - '-n', '--resource-name', - metavar='NAME', - type=str, - dest='rname', - required=True, - help='The name of the BEINC-resource on the remote server') - parser.add_argument( - '-o', '--osd-system', - metavar='SYSTEM', - type=str, - dest='osd_sys', - default='pynotify', - help='BEINC osd-system: "pynotify" (default)') - parser.add_argument( - '-p', '--password', - metavar='PASSWORD[FILE]', - type=str, - dest='password', - default='', - help='BEINC taget-password / text-file containing the target password' - ' (default & recommended: prompt for passwd)') - parser.add_argument( - '-T', '--socket-timeout', - metavar='SECONDS', - type=int, - dest='socket_timeout', - default=3, - help='Socket timeout in seconds (0=Python default) (default: 3)') - parser.add_argument( - '-t', '--osd-timeout', - metavar='SECONDS', - type=int, - dest='osd_timeout', - default=5, - help='OSD timeout (default: 5)') - parser.add_argument( - '-v', '--version', - action='version', - version='%(prog)s {0}'.format(__version__), - help='display program-version and exit') - args = parser.parse_args() - if not args.password: - try: - args.password = getpass.getpass() - except Exception as e: - sys.stderr.write('Prompt terminated\n') - sys.exit(errno.EACCES) - elif os.path.isfile(args.password): - try: - with open(args.password, 'r') as fp: - passwd = fp.readline() - if passwd.strip(): - args.password = passwd.strip() - except Exception as e: - sys.stderr.write('Unable to open password file: {0}'.format(e)) - sys.exit(1) - scheduler = sched.scheduler(time.time, time.sleep) - scheduler.enter(args.frequency, - 1, - poll_notifications, - (scheduler, args)) - scheduler.run() - - -if __name__ == '__main__': - main() diff --git a/beinc_pull.py b/beinc_pull.py new file mode 100755 index 0000000..9c44c00 --- /dev/null +++ b/beinc_pull.py @@ -0,0 +1,248 @@ +#!/usr/bin/env python +# -*- coding: utf-8 -*- + +# Blackmore's Enhanced IRC-Notification Collection (BEINC) v4.0 +# Copyright (C) 2013-2020 Simeon Simeonov + +# 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 . +"""A simple client that pulls notifications from a BEINC server""" +import argparse +import errno +import getpass +import io +import json +import os +import sched +import socket +import ssl +import sys +import time +import urllib.parse +import urllib.request + +try: + import notify2 as pynotify +except ImportError: + pynotify = None + + +__author__ = 'Simeon Simeonov' +__version__ = '4.0' +__license__ = 'GPL3' + + +def eprint(*arg, **kwargs): + """stdderr print wrapper""" + print(*arg, file=sys.stderr, flush=True, **kwargs) + + +def fetch_password(args_password): + """ + Fetches the password from the provided `args_password` + + :param args_password: The password coming from argparse + :type args_password: str + + :return: The password string + :rtype: str + """ + if not args_password: + try: + return getpass.getpass() + except KeyboardInterrupt: + eprint(os.linesep + 'Prompt terminated') + sys.exit(errno.EACCES) + elif os.path.isfile(args_password): + try: + with io.open(args_password, 'r') as fp: + passwd = fp.readline() + if passwd.strip(): + return passwd.strip() + except Exception as e: + eprint(f'Unable to open password file: {e}') + sys.exit(1) + return args_password + + +def display_notification(args, title, message): + """ + A wrapper function for displaying a single notification + + :param args: The arguments assigned from argparse + :type args: argparse.Namespace + + :param title: The title + :type title: str + + :param message: The message + :type message: str + """ + if args.osd_sys == 'pynotify': + if not pynotify: + raise Exception( + 'Could not load "pynotify".' + 'Please install "pynotify" or use a different osd-system!' + 'Terminating...') + if not pynotify.init('BEINC Notify'): + raise Exception('There was a problem with libnotify') + notification_obj = pynotify.Notification(summary=title, + message=message) + notification_obj.timeout = 1000 * args.osd_timeout + notification_obj.set_category('im.received') + notification_obj.show() + else: + raise Exception(f'Unsupported osd-system: {args.osd_sys}') + + +def pull_notifications(scheduler, args): + """ + The core function initiated by the scheduler performing a single pull + + :param scheduler: Scheduler object + :type scheduler: sched.scheduler + + :param args: The arguments assigned from argparse + :type args: argparse.Namespace + """ + try: + context = ssl.SSLContext(ssl.PROTOCOL_TLSv1_2) + context.verify_mode = ssl.CERT_NONE + if args.cert: + context.verify_mode = ssl.CERT_REQUIRED + context.load_verify_locations(cafile=os.path.expanduser(args.cert)) + context.check_hostname = bool(not args.disable_hostname_check) + if args.ciphers: + context.set_ciphers(args.ciphers) + response = urllib.request.urlopen( + args.url, + data=urllib.parse.urlencode( + ( + ('resource_name', args.rname), + ('password', args.password) + )).encode('utf-8'), + timeout=args.socket_timeout, + context=context) + response_dict = json.loads(response.read().decode('utf-8')) + if response.code != 200: + raise socket.error(response_dict.get('message', '')) + for entry in response_dict['data']['messages']: + display_notification(args, + entry.get('title', ''), + entry.get('message', '')) + response.close() + scheduler.enter(args.frequency, + 1, + pull_notifications, + (scheduler, args)) + except ssl.SSLError as e: + eprint(f'BEINC SSL/TLS error: {e}') + sys.exit(errno.EPERM) + except socket.error as e: + eprint(f'BEINC connection error: {e}') + sys.exit(errno.EPERM) + except Exception as e: + eprint(f'BEINC generic client error: {e}') + sys.exit(errno.EPERM) + + +def main(inargs=None): + """main entry""" + parser = argparse.ArgumentParser( + description='The following options are available') + parser.add_argument( + 'url', + metavar='URL', + type=str, + help='BEINC server destination URL') + parser.add_argument( + '-c', '--cert-file', + metavar='FILE', + type=str, + dest='cert', + default='', + help='CA-cert to check the server-cert against ' + '(default: Check disabled)') + parser.add_argument( + '--ciphers', + metavar='CIPHERS', + type=str, + dest='ciphers', + default='', + help='Preferred ciphers list (default: auto)') + parser.add_argument( + '--disable-hostname-check', + action='store_true', + dest='disable_hostname_check', + default=False, + help='Do not check whether server cert matches server hostname') + parser.add_argument( + '-f', '--frequency', + metavar='SECONDS', + type=int, + dest='frequency', + default=10, + help='Pulling frequency in seconds (default: 10)') + parser.add_argument( + '-n', '--resource-name', + metavar='NAME', + type=str, + dest='rname', + required=True, + help='The name of the BEINC-resource on the remote server') + parser.add_argument( + '-o', '--osd-system', + metavar='SYSTEM', + type=str, + dest='osd_sys', + default='pynotify', + help='BEINC osd-system: "pynotify" (default)') + parser.add_argument( + '-p', '--password', + metavar='PASSWORD[FILE]', + type=str, + dest='password', + default='', + help='BEINC taget-password / text-file containing the target password' + ' (default & recommended: prompt for passwd)') + parser.add_argument( + '-T', '--socket-timeout', + metavar='SECONDS', + type=int, + dest='socket_timeout', + default=3, + help='Socket timeout in seconds (0=Python default) (default: 3)') + parser.add_argument( + '-t', '--osd-timeout', + metavar='SECONDS', + type=int, + dest='osd_timeout', + default=5, + help='OSD timeout (default: 5)') + parser.add_argument( + '-v', '--version', + action='version', + version=f'%(prog)s {__version__}', + help='display program-version and exit') + args = parser.parse_args(inargs) + args.password = fetch_password(args.password) + scheduler = sched.scheduler(time.time, time.sleep) + scheduler.enter(args.frequency, + 1, + pull_notifications, + (scheduler, args)) + scheduler.run() + + +if __name__ == '__main__': + main() diff --git a/beinc_server.py b/beinc_server.py index ad81ce9..0f6c248 100755 --- a/beinc_server.py +++ b/beinc_server.py @@ -1,8 +1,8 @@ #!/usr/bin/env python # -*- coding: utf-8 -*- -# Blackmore's Enhanced IRC-Notification Collection (BEINC) v3.0 -# Copyright (C) 2013-2018 Simeon Simeonov +# Blackmore's Enhanced IRC-Notification Collection (BEINC) v4.0 +# Copyright (C) 2013-2020 Simeon Simeonov # 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 @@ -29,26 +29,16 @@ import sys from functools import wraps from logging.config import fileConfig - -PY2 = sys.version_info[0] == 2 -PY3 = sys.version_info[0] == 3 - -if PY3: - from http.server import BaseHTTPRequestHandler, HTTPServer -else: - from BaseHTTPServer import BaseHTTPRequestHandler, HTTPServer +from http.server import BaseHTTPRequestHandler, HTTPServer try: - if PY3: - import notify2 as pynotify - else: - import pynotify -except ImportError as e: + import notify2 as pynotify +except ImportError: pynotify = None __author__ = 'Simeon Simeonov' -__version__ = '3.0' +__version__ = '4.0' __license__ = 'GPL3' @@ -59,28 +49,30 @@ BEINC_CURRENT_CONFIG_VERSION = 3 class BEINCError401(Exception): - pass + """BEINCError401""" class BEINCError403(Exception): - pass + """BEINCError403""" class BEINCError404(Exception): - pass + """BEINCError404""" class BEINCError405(Exception): - pass + """BEINCError405""" -def beinc_login_required(method): - """ - Decorator for checking login credentials - """ +def eprint(*arg, **kwargs): + """stdderr print wrapper""" + print(*arg, file=sys.stderr, flush=True, **kwargs) + +def beinc_login_required(method): + """Decorator for checking login credentials""" @wraps(method) - def wrapper(self, data, *args, **kwargs): + def wrapper(self, data, *arg, **kwargs): if data.get('resource_name') is None: raise BEINCError403('Resource-name missing') if data.get('password') is None: @@ -91,122 +83,122 @@ def beinc_login_required(method): raise BEINCError401('Wrong instance or password') if not instance.password_match(data.get('password')): raise BEINCError401('Wrong instance or password') - return method(self, data, *args, **kwargs) + return method(self, data, *arg, **kwargs) return wrapper -class BEINCInstance(object): - """ - Represents a single server-instance - """ +class BEINCInstance: + """Represents a single server-instance""" def __init__(self, instance_dict): """ instance_dict: the config-dictionary node that represents this instance """ - self.__message_queue = list() - self.__osd_type = BEINC_OSD_TYPE_NONE - self.__osd_notification = None + self._message_queue = [] + self._osd_type = BEINC_OSD_TYPE_NONE + self._osd_notification = None - self.__name = instance_dict.get('name') - self.__password = instance_dict.get('password', '') - self.__queue_size = int(instance_dict.get('queue_size', 3)) + self._name = instance_dict.get('name') + self._password = instance_dict.get('password', '') + self._queue_size = int(instance_dict.get('queue_size', 3)) if instance_dict['osd_system'].lower() == 'pynotify': - self.__queue_size = 0 # disable queueing + self._queue_size = 0 # disable queueing if pynotify is None: - sys.stderr.write( - 'This server does not possess pynotify capability\n') - sys.stderr.write( - 'Remove the instance {0}'.format(self.__name)) - sys.stderr.write( - 'or define it with "osd_system": "none" ' - 'or other available backend\n') + eprint('This server does not possess pynotify capability') + eprint(f'Remove the instance {self._name} or define it with ' + f'"osd_system": "none" or other ' + f'available backend') sys.exit(errno.EPERM) try: - self.__osd_notification = pynotify.Notification(' ') - if PY3: - self.__osd_notification.timeout = 1000 * int( - instance_dict.get('osd_timeout', 5)) - self.__osd_notification.set_category('im.received') - else: - self.__osd_notification.set_timeout( - 1000 * int(instance_dict.get('osd_timeout', 5))) - self.__osd_notification.set_property( - 'app_name', - '{0} {1}'.format(sys.argv[0], __version__)) - self.__osd_type = BEINC_OSD_TYPE_PYNOTIFY + self._osd_notification = pynotify.Notification(' ') + self._osd_notification.timeout = 1000 * int( + instance_dict.get('osd_timeout', 5)) + self._osd_notification.set_category('im.received') + self._osd_type = BEINC_OSD_TYPE_PYNOTIFY except Exception as e: - sys.stderr.write( - 'Unable to set up a ' - 'pynotify notification object for "{0}" ({1})\n'.format( - self.__name, - e)) + eprint(f'Unable to set up a pynotify notification object ' + f'for "{self._name}" ({e})') sys.exit(errno.EPERM) @property def name(self): - """ - name-property for the server instance (read-only) - """ - return self.__name + """name-property for the server instance (read-only)""" + return self._name @property def queueable(self): - """ - True if this instance has a queueing capability (read-only) - """ - return bool(self.__queue_size) + """True if this instance has a queueing capability (read-only)""" + return bool(self._queue_size) def password_match(self, password): """ Returns True if 'passowrd' matches the instance-password, otherwise - False + + :param password: The password to compare + :type password: str + + :return: True if the password matches, False otherwise + :rtype: bool """ - return True if self.__password == password else False + return self._password == password def send_message(self, title, message): """ Displays or enqueues the message, depending on the instance's type in regard to the osd_system + + :param title: The title + :type title: str + + :param message: The message + :type message: str """ - if self.__osd_type == BEINC_OSD_TYPE_PYNOTIFY: - self.__send_pynotify_messaage(title, message) + if self._osd_type == BEINC_OSD_TYPE_PYNOTIFY: + self._send_pynotify_messaage(title, message) else: - self.__send_message_to_queue(title, message) + self._send_message_to_queue(title, message) def get_queue(self): - """ - Returns a list of dict representation of the message queue - """ - r_value = self.__message_queue - self.__message_queue = list() + """Returns a list of dict representation of the message queue""" + r_value = self._message_queue + self._message_queue = [] return r_value - def __send_pynotify_messaage(self, title, message): + def _send_pynotify_messaage(self, title, message): """ Displays pynotify message + + :param title: The title + :type title: str + + :param message: The message + :type message: str """ - self.__osd_notification.update(summary=title, message=message) - self.__osd_notification.show() + self._osd_notification.update(summary=title, message=message) + self._osd_notification.show() - def __send_message_to_queue(self, title, message): + def _send_message_to_queue(self, title, message): """ Enqueues the message + + :param title: The title + :type title: str + + :param message: The message + :type message: str """ - if len(self.__message_queue) >= self.__queue_size: - self.__message_queue.pop(0) - self.__message_queue.append({'title': title, 'message': message}) + if len(self._message_queue) >= self._queue_size: + self._message_queue.pop(0) + self._message_queue.append({'title': title, 'message': message}) class BEINCCustomHandler(BaseHTTPRequestHandler): - """ - """ + """Custom handler""" def do_POST(self): - """ - Handle POST requests - """ + """Handle POST requests""" if self.path.strip('/') not in ('beinc/push', 'beinc/pull'): - self.__generate_json_error(404, 'Invalid resource path') + self._generate_json_error(404, 'Invalid resource path') return form = cgi.FieldStorage( fp=self.rfile, @@ -222,43 +214,38 @@ class BEINCCustomHandler(BaseHTTPRequestHandler): try: result = {} if self.path.strip('/') == 'beinc/push': - result = self.__handle_push(POST_data) + result = self._handle_push(POST_data) elif self.path.strip('/') == 'beinc/pull': - result = self.__handle_pull(POST_data) - self.__render_to_JSON_response(result) + result = self._handle_pull(POST_data) + self._render_to_JSON_response(result) except BEINCError401 as e: - self.__generate_json_error(401, str(e)) + self._generate_json_error(401, str(e)) except BEINCError403 as e: - self.__generate_json_error(403, str(e)) + self._generate_json_error(403, str(e)) except BEINCError404 as e: - self.__generate_json_error(404, str(e)) + self._generate_json_error(404, str(e)) except BEINCError405 as e: - self.__generate_json_error(405, str(e)) + self._generate_json_error(405, str(e)) except Exception as e: - self.__generate_json_error(500, - 'Unexpected error: {}'.format(str(e))) + self._generate_json_error(500, f'Unexpected error: {e}') def do_GET(self): - """ - Handle GET Requests - """ - self.__generate_json_error(405, 'Unsupported method') + """Handle GET Requests""" + self._generate_json_error(405, 'Unsupported method') @beinc_login_required - def __handle_push(self, data): - """ - """ + def _handle_push(self, data): + """Handle push""" instance = self.server.instances[data.get('resource_name')] try: instance.send_message(data.get('title'), data.get('message')) return {'message': 'OK. Sent.'} except Exception as e: - self.__generate_json_error(500, str(e)) + self._generate_json_error(500, str(e)) @beinc_login_required - def __handle_pull(self, data): - """ - """ + def _handle_pull(self, data): + """Handle pull""" instance = self.server.instances[data.get('resource_name')] try: if not instance.queueable: @@ -267,13 +254,12 @@ class BEINCCustomHandler(BaseHTTPRequestHandler): return {'message': 'OK. Fetched.', 'data': {'messages': instance.get_queue()}} except Exception as e: - self.__generate_json_error(500, str(e)) + self._generate_json_error(500, str(e)) - def __generate_json_error(self, code, message): + def _generate_json_error(self, code, message): """ Generates response header and json content for errors - Keyword Arguments: :param code: the HTTP code :type code: int @@ -288,9 +274,8 @@ class BEINCCustomHandler(BaseHTTPRequestHandler): sort_keys=True, indent=4).encode('utf-8')) - def __render_to_JSON_response(self, context): + def _render_to_JSON_response(self, context): """ - Keyword Arguments: :param context: the context-dict to be converted to json :type context: dict """ @@ -305,31 +290,36 @@ class BEINCCustomHandler(BaseHTTPRequestHandler): class BEINCNotifyServer(HTTPServer): - """ - """ + """BEINCNotifyServer class""" + def __init__(self, *arg, **kwargs): + """Default constructor""" + super().__init__(*arg, **kwargs) + self._config = None + self._instances = {} + def set_config(self, config): """ Sets the configuration dict for the server, instantiates the BEINC instances and initiates the defined OSD backends """ - self.__config = config - self.__instances = dict() + self._config = config # initialize pynotify if the module exists and if needed if pynotify: - for instance in self.__config['server']['instances']: + for instance in self._config['server']['instances']: # check if we have at least one instance that uses pynotify # before initializing it if instance.get('osd_system', '').lower() == 'pynotify': if not pynotify.init('BEINC Notify'): - sys.stderr.write('pynotify.init failed! Exiting...\n') + eprint('pynotify.init failed! Exiting...') sys.exit(1) break + instance = {'name': 'Invalid'} try: - for instance in self.__config['server']['instances']: - self.__instances[instance['name']] = BEINCInstance(instance) - logger.info('Instance "{0}" added'.format(instance['name'])) + for instance in self._config['server']['instances']: + self._instances[instance['name']] = BEINCInstance(instance) + logger.info('Instance %s added', instance['name']) except Exception as e: - sys.stderr.write('Unable to create instance "{0}": {1}\n'.format( + eprint('Unable to create instance "{0}": {1}'.format( instance['name'], e)) sys.exit(1) @@ -339,7 +329,7 @@ class BEINCNotifyServer(HTTPServer): """ a property that returns the instance list (read-only) """ - return self.__instances + return self._instances if __name__ == '__main__': @@ -390,15 +380,14 @@ if __name__ == '__main__': parser.add_argument( '-v', '--version', action='version', - version='%(prog)s {0}'.format(__version__), + version=f'%(prog)s {__version__}', help='Display program-version and exit') args = parser.parse_args() try: with open(args.config_file, 'r') as fp: config_dict = json.load(fp) except Exception as e: - sys.stderr.write('Unable to parse {0}: {1}\n'.format(args.config_file, - e)) + eprint(f'Unable to parse {args.config_file}: {e}') sys.exit(errno.EIO) try: if os.path.isfile(args.logger_config): @@ -411,11 +400,11 @@ if __name__ == '__main__': logger = logging.getLogger('beinc') logger.info('BEINC starting. Loading config...') if config_dict.get('config_version') != BEINC_CURRENT_CONFIG_VERSION: - sys.stderr.write( + eprint( 'WARNING: The version of the config-file: {0} ({1}) ' 'does not correspond to the latest version supported ' 'by this program ({2})\nCheck beinc_config_sample.json ' - 'for the newest features!\n'.format( + 'for the newest features!'.format( args.config_file, config_dict.get('config_version', 'Not set'), BEINC_CURRENT_CONFIG_VERSION)) @@ -441,6 +430,6 @@ if __name__ == '__main__': except KeyboardInterrupt: print('\n\nTerminating...') except Exception as e: - sys.stderr.write('BEINCServer critical error: {0}\n'.format(e)) + eprint(f'BEINCServer critical error: {e}') sys.exit(1) sys.exit(0) diff --git a/beinc_weechat.py b/beinc_weechat.py index 132a22a..c1a1e1e 100644 --- a/beinc_weechat.py +++ b/beinc_weechat.py @@ -1,7 +1,7 @@ # -*- coding: utf-8 -*- -# Blackmore's Enhanced IRC-Notification Collection (BEINC) v3.0 -# Copyright (C) 2013-2019 Simeon Simeonov +# Blackmore's Enhanced IRC-Notification Collection (BEINC) v4.0 +# Copyright (C) 2013-2020 Simeon Simeonov # 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 @@ -15,35 +15,25 @@ # You should have received a copy of the GNU General Public License # along with this program. If not, see . - - +"""BEINC client for Weechat""" import datetime import json import os import socket import ssl -import sys +import urllib.parse +import urllib.request import weechat -PY2 = sys.version_info[0] == 2 -PY3 = sys.version_info[0] == 3 - -if PY3: - from urllib.parse import urlencode - from urllib.request import urlopen -else: - from urllib import urlencode - from urllib2 import urlopen - __author__ = 'Simeon Simeonov' -__version__ = '3.0' +__version__ = '4.0' __license__ = 'GPL3' enabled = True -global_values = dict() +global_values = {} # few constants # BEINC_POLICY_NONE = 0 @@ -52,322 +42,317 @@ BEINC_POLICY_LIST_ONLY = 2 BEINC_CURRENT_CONFIG_VERSION = 2 -class WeechatTarget(object): +class WeechatTarget: """ The target (destination) class + Each remote destination is represented as a WeechatTarget object """ def __init__(self, target_dict): """ - target_dict: the config-dictionary node that represents this instance + :param target_dict: The config-dict node that represents this instance + :type target_dict: dict """ - self.__name = target_dict.get('name', '') - if self.__name == '': + self._name = target_dict.get('name', '') + if self._name == '': raise Exception('"name" not defined for target') - self.__url = target_dict.get('target_url', '') - if self.__url == '': + self._url = target_dict.get('target_url', '') + if self._url == '': raise Exception('"target_url" not defined for target') - self.__password = target_dict.get('target_password', '') - self.__pm_title_template = target_dict.get('pm_title_template', - '%s @ %S') - self.__pm_message_template = target_dict.get('pm_message_template', - '%m') - self.__cm_title_template = target_dict.get('cm_title_template', - '%c @ %S') - self.__cm_message_template = target_dict.get('cm_message_template', - '%s -> %m') - self.__nm_title_template = target_dict.get('nm_title_template', - '%c @ %S') - self.__nm_message_template = target_dict.get('nm_message_template', - '%s -> %m') - self.__chans = set(target_dict.get('channel_list', list())) - self.__nicks = set(target_dict.get('nick_list', list())) - self.__chan_messages_policy = int(target_dict.get( + self._password = target_dict.get('target_password', '') + self._pm_title_template = target_dict.get('pm_title_template', + '%s @ %S') + self._pm_message_template = target_dict.get('pm_message_template', + '%m') + self._cm_title_template = target_dict.get('cm_title_template', + '%c @ %S') + self._cm_message_template = target_dict.get('cm_message_template', + '%s -> %m') + self._nm_title_template = target_dict.get('nm_title_template', + '%c @ %S') + self._nm_message_template = target_dict.get('nm_message_template', + '%s -> %m') + self._chans = set(target_dict.get('channel_list', [])) + self._nicks = set(target_dict.get('nick_list', [])) + self._chan_messages_policy = int(target_dict.get( 'channel_messages_policy', BEINC_POLICY_LIST_ONLY)) - self.__priv_messages_policy = int(target_dict.get( + self._priv_messages_policy = int(target_dict.get( 'private_messages_policy', BEINC_POLICY_ALL)) - self.__notifications_policy = int(target_dict.get( + self._notifications_policy = int(target_dict.get( 'notifications_policy', BEINC_POLICY_ALL)) - self.__cert_file = target_dict.get('target_cert_file') - self.__timestamp_format = target_dict.get('target_timestamp_format', - '%H:%M:%S') - self.__debug = bool(target_dict.get('debug', False)) - self.__enabled = bool(target_dict.get('enabled', True)) - self.__socket_timeout = int(target_dict.get('socket_timeout', 3)) - self.__ssl_ciphers = target_dict.get('ssl_ciphers', '') - self.__disable_hostname_check = bool( + self._cert_file = target_dict.get('target_cert_file') + self._timestamp_format = target_dict.get('target_timestamp_format', + '%H:%M:%S') + self._debug = bool(target_dict.get('debug', False)) + self._enabled = bool(target_dict.get('enabled', True)) + self._socket_timeout = int(target_dict.get('socket_timeout', 3)) + self._ssl_ciphers = target_dict.get('ssl_ciphers', '') + self._disable_hostname_check = bool( target_dict.get('disable-hostname-check', False)) - self.__ssl_version = target_dict.get('ssl_version', 'auto') - self.__last_message = None # datetime.datetime instance - self.__context = None - self.__context_setup() + self._ssl_version = target_dict.get('ssl_version', 'auto') + self._last_message = None # datetime.datetime instance + self._context = None + self._context_setup() @property def name(self): - """ - Target name (read-only property) - """ - return self.__name + """Target name (read-only property)""" + return self._name @property def chans(self): - """ - Target channel list (read-only property) - """ - return self.__chans + """Target channel list (read-only property)""" + return self._chans @property def nicks(self): - """ - Target nick list (read-only property) - """ - return self.__nicks + """Target nick list (read-only property)""" + return self._nicks @property def channel_messages_policy(self): - """ - The target's channel messages policy (read-only property) - """ - return self.__chan_messages_policy + """The target's channel messages policy (read-only property)""" + return self._chan_messages_policy @property def private_messages_policy(self): - """ - The target's private messages policy (read-only property) - """ - return self.__priv_messages_policy + """The target's private messages policy (read-only property)""" + return self._priv_messages_policy @property def notifications_policy(self): - """ - The target's notifications policy (read-only property) - """ - return self.__notifications_policy + """The target's notifications policy (read-only property)""" + return self._notifications_policy @property def enabled(self): - """ - The target's enabled status (bool property) - """ - return self.__enabled + """The target's enabled status (bool property)""" + return self._enabled @enabled.setter def enabled(self, value): - """ - The target's enabled status (bool property) - """ - self.__enabled = value + """The target's enabled status (bool property)""" + self._enabled = value def __repr__(self): - """ - """ + """repr() implementation""" last_message = 'never' - if self.__last_message is not None: - last_message = self.__last_message.strftime('%Y-%m-%d %H:%M:%S') + if self._last_message is not None: + last_message = self._last_message.strftime('%Y-%m-%d %H:%M:%S') return ('name: {0}\nurl: {1}\nenabled: {2}\nchannel_list: {3}\n' 'nick_list: {4}\nchannel_messages_policy: {5}\n' 'private_messages_policy: {6}\nnotifications_policy: {7}\n' 'last message: {8}\nsocket timeout: {9}\nssl-version: {10}\n' 'ciphers: {11}\ndisable hostname check: {12}\n' 'debug: {13}\n\n'.format( - self.__name, - self.__url, - 'yes' if self.__enabled else 'no', - ', '.join(self.__chans), - ', '.join(self.__nicks), - self.__chan_messages_policy, - self.__priv_messages_policy, - self.__notifications_policy, + self._name, + self._url, + 'yes' if self._enabled else 'no', + ', '.join(self._chans), + ', '.join(self._nicks), + self._chan_messages_policy, + self._priv_messages_policy, + self._notifications_policy, last_message, - self.__socket_timeout, - self.__ssl_version, - self.__ssl_ciphers or 'auto', - 'yes' if self.__disable_hostname_check else 'no', - 'yes' if self.__debug else 'no')) + self._socket_timeout, + self._ssl_version, + self._ssl_ciphers or 'auto', + 'yes' if self._disable_hostname_check else 'no', + 'yes' if self._debug else 'no')) def send_private_message_notification(self, values): """ - sends a private message notification to the represented target + Sends a private message notification to the represented target - values: dict pupulated by the irc msg-handler + :param value: Dict pupulated by the irc msg-handler + :type value: dict """ try: - title = self.__fetch_formatted_str(self.__pm_title_template, - values) - message = self.__fetch_formatted_str( - self.__pm_message_template, + title = self._fetch_formatted_str(self._pm_title_template, + values) + message = self._fetch_formatted_str( + self._pm_message_template, values) - if not self.__send_beinc_message(title, message) and self.__debug: + if not self._send_beinc_message(title, message) and self._debug: beinc_prnt( - 'BEINC DEBUG: send_private_message_notification-ERROR ' - 'for "{0}": __send_beinc_message -> False'.format( - self.__name)) + f'BEINC DEBUG: send_private_message_notification-ERROR ' + f'for "{self._name}": _send_beinc_message -> False') except Exception as e: - if self.__debug: + if self._debug: beinc_prnt( - 'BEINC DEBUG: send_private_message_notification-ERROR ' - 'for "{0}": {1}'.format(self.__name, e)) + f'BEINC DEBUG: send_private_message_notification-ERROR ' + f'for "{self._name}": {e}') def send_channel_message_notification(self, values): """ - sends a channel message notification to the represented target + Sends a channel message notification to the represented target - values: dict pupulated by the irc msg-handler + :param value: Dict pupulated by the irc msg-handler + :type value: dict """ try: - title = self.__fetch_formatted_str(self.__cm_title_template, - values) - message = self.__fetch_formatted_str(self.__cm_message_template, - values) - if not self.__send_beinc_message(title, message) and self.__debug: + title = self._fetch_formatted_str(self._cm_title_template, + values) + message = self._fetch_formatted_str(self._cm_message_template, + values) + if not self._send_beinc_message(title, message) and self._debug: beinc_prnt( - 'BEINC DEBUG: send_channel_message_notification-ERROR ' - 'for "{0}": __send_beinc_message -> False'.format( - self.__name)) + f'BEINC DEBUG: send_channel_message_notification-ERROR ' + f'for "{self._name}": _send_beinc_message -> False') except Exception as e: - if self.__debug: + if self._debug: beinc_prnt( - 'BEINC DEBUG: send_channel_message_notification-ERROR ' - 'for "{0}": {1}'.format(self.__name, e)) + f'BEINC DEBUG: send_channel_message_notification-ERROR ' + f'for "{self._name}": {e}') def send_notify_message_notification(self, values): """ - sends a notify message notification to the represented target + Sends a notify message notification to the represented target - values: dict pupulated by the irc msg-handler + :param value: Dict pupulated by the irc msg-handler + :type value: dict """ try: - title = self.__fetch_formatted_str(self.__nm_title_template, - values) - message = self.__fetch_formatted_str(self.__nm_message_template, - values) - if not self.__send_beinc_message(title, message) and self.__debug: + title = self._fetch_formatted_str(self._nm_title_template, + values) + message = self._fetch_formatted_str(self._nm_message_template, + values) + if not self._send_beinc_message(title, message) and self._debug: beinc_prnt( - 'BEINC DEBUG: send_notify_message_notification-ERROR ' - 'for "{0}": __send_beinc_message -> False'.format( - self.__name)) + f'BEINC DEBUG: send_notify_message_notification-ERROR ' + f'for "{self._name}": _send_beinc_message -> False') except Exception as e: - if self.__debug: + if self._debug: beinc_prnt( - 'BEINC DEBUG: send_notify_message_notification-ERROR ' - 'for "{0}": {1}'.format(self.__name, e)) + f'BEINC DEBUG: send_notify_message_notification-ERROR ' + f'for "{self._name}": {e}') def send_broadcast_notification(self, message): """ - sends a 'pure' broadcast / test message notification + Sends a 'pure' broadcast / test message notification to the represented target - message: a single message string + :param message: A single message string + :type message: str """ try: title = 'BEINC broadcast' - if not self.__send_beinc_message(title, message) and self.__debug: + if not self._send_beinc_message(title, message) and self._debug: beinc_prnt( - 'BEINC DEBUG: send_broadcast_notification-ERROR ' - 'for "{0}": __send_beinc_message -> False'.format( - self.__name)) + f'BEINC DEBUG: send_broadcast_notification-ERROR ' + f'for "{self._name}": _send_beinc_message -> False') except Exception as e: - if self.__debug: + if self._debug: beinc_prnt( - 'BEINC DEBUG: send_broadcast_notification-ERROR ' - 'for "{0}": {1}'.format(self.__name, e)) + f'BEINC DEBUG: send_broadcast_notification-ERROR ' + f'for "{self._name}": {e}') - def __context_setup(self): - """ - """ - if self.__context is not None: + def _context_setup(self): + """Sets up the SSL context""" + if self._context is not None: return True try: context = ssl.SSLContext(ssl.PROTOCOL_TLSv1_2) context.verify_mode = ssl.CERT_NONE - if self.__cert_file: + if self._cert_file: context.verify_mode = ssl.CERT_REQUIRED context.load_verify_locations(cafile=os.path.expanduser( - self.__cert_file)) + self._cert_file)) context.check_hostname = bool( - not self.__disable_hostname_check) - if self.__ssl_ciphers and self.__ssl_ciphers != 'auto': - context.set_ciphers(self.__ssl_ciphers) - self.__context = context + not self._disable_hostname_check) + if self._ssl_ciphers and self._ssl_ciphers != 'auto': + context.set_ciphers(self._ssl_ciphers) + self._context = context return True except ssl.SSLError as e: - if self.__debug: - beinc_prnt('BEINC DEBUG: SSL/TLS error: {0}\n'.format(e)) + if self._debug: + beinc_prnt(f'BEINC DEBUG: SSL/TLS error: {e}\n') except Exception as e: - if self.__debug: - beinc_prnt('BEINC DEBUG: Generic context error: {0}\n'.format( - e)) - self.__context = None + if self._debug: + beinc_prnt(f'BEINC DEBUG: Generic context error: {e}\n') + self._context = None return False - def __fetch_formatted_str(self, template, values): + def _fetch_formatted_str(self, template, values): """ - returns a formatted string by replacing the defined + Returns a formatted string by replacing the defined macros in 'template' the the corresponding values from 'values' - values: dict - template: str + :param template: The template to use + :type template: str + + :param values: The values dict + :type values: dict + + :return: The formatted string + :rtype: str """ - timestamp = datetime.datetime.now().strftime(self.__timestamp_format) + timestamp = datetime.datetime.now().strftime(self._timestamp_format) replacements = {'%S': values['server'], '%s': values['source_nick'], '%c': values['channel'], '%m': values['message'], '%t': timestamp, - '%p': u'BEINC', + '%p': 'BEINC', '%n': values['own_nick']} for key, value in replacements.items(): template = template.replace(key, value) - return template.encode('utf-8') + return template - def __send_beinc_message(self, title, message): + def _send_beinc_message(self, title, message): """ - the function implements the BEINC "protocol" by generating a simple + The method implements the BEINC "protocol" by generating a simple POST request + + :param title: The title + :type title: str + + :param message: The message + :type message: str + + :return: The status + :rtype: bool """ try: - if self.__context is None and not self.__context_setup(): + if self._context is None and not self._context_setup(): return False - response = urlopen( - self.__url, - data=urlencode( + response = urllib.request.urlopen( + self._url, + data=urllib.parse.urlencode( ( - ('resource_name', self.__name), - ('password', self.__password), + ('resource_name', self._name), + ('password', self._password), ('title', title), ('message', message) )).encode('utf-8'), - timeout=self.__socket_timeout, - context=self.__context) + timeout=self._socket_timeout, + context=self._context) response_dict = json.loads(response.read().decode('utf-8')) if response.code != 200: raise socket.error(response_dict.get('message', '')) - if self.__debug: + if self._debug: beinc_prnt('BEINC DEBUG: Server responded: {0}'.format( response_dict.get('message'))) - self.__last_message = datetime.datetime.now() + self._last_message = datetime.datetime.now() return True except ssl.SSLError as e: - if self.__debug: - beinc_prnt('BEINC DEBUG: SSL/TLS error: {0}\n'.format(e)) + if self._debug: + beinc_prnt(f'BEINC DEBUG: SSL/TLS error: {e}\n') except socket.error as e: - if self.__debug: - beinc_prnt('BEINC DEBUG: Connection error: {0}\n'.format(e)) + if self._debug: + beinc_prnt(f'BEINC DEBUG: Connection error: {e}\n') except Exception as e: - if self.__debug: - beinc_prnt('BEINC DEBUG: Unable to send message: {0}\n'.format( - e)) + if self._debug: + beinc_prnt(f'BEINC DEBUG: Unable to send message: {e}\n') return False def beinc_prnt(message_str): - """ - wrapper around weechat.prnt - """ + """wrapper around weechat.prnt""" if global_values['use_current_buffer']: weechat.prnt(weechat.current_buffer(), message_str) else: @@ -375,9 +360,7 @@ def beinc_prnt(message_str): def beinc_cmd_broadcast_handler(cmd_tokens): - """ - handles: '/beinc broadcast' command actions - """ + """handles: '/beinc broadcast' command actions""" if not cmd_tokens: beinc_prnt('beinc broadcast ') return weechat.WEECHAT_RC_OK @@ -388,9 +371,7 @@ def beinc_cmd_broadcast_handler(cmd_tokens): def beinc_cmd_target_handler(cmd_tokens): - """ - handles: '/beinc target' command actions - """ + """handles: '/beinc target' command actions""" if not cmd_tokens or cmd_tokens[0] not in ['list', 'enable', 'disable']: beinc_prnt('beinc target [ list | enable | disable ]') return weechat.WEECHAT_RC_OK @@ -410,10 +391,10 @@ def beinc_cmd_target_handler(cmd_tokens): for target in target_list: if target.name == name: target.enabled = True - beinc_prnt('target "{0}" enabled'.format(name)) + beinc_prnt(f'target "{name}" enabled') break else: - beinc_prnt('no matching target for "{0}"'.format(name)) + beinc_prnt(f'no matching target for "{name}"') elif cmd_tokens[0] == 'disable': if not cmd_tokens[1:]: beinc_prnt('missing a name-argument') @@ -422,17 +403,15 @@ def beinc_cmd_target_handler(cmd_tokens): for target in target_list: if target.name == name: target.enabled = False - beinc_prnt('target "{0}" disabled'.format(name)) + beinc_prnt(f'target "{name}" disabled') break else: - beinc_prnt('no matching target for "{0}"'.format(name)) + beinc_prnt(f'no matching target for "{name}"') return weechat.WEECHAT_RC_OK def beinc_command(data, buffer_obj, args): - """ - Callback function handling the Weechat's /beinc command - """ + """Callback function handling the Weechat's /beinc command""" global enabled cmd_tokens = args.split() if not cmd_tokens: @@ -457,15 +436,13 @@ def beinc_command(data, buffer_obj, args): def beinc_privmsg_handler(data, signal, signal_data): - """ - Callback function the *PRIVMSG* IRC messages hooked by Weechat - """ + """Callback function the *PRIVMSG* IRC messages hooked by Weechat""" if not enabled: return weechat.WEECHAT_RC_OK prvmsg_dict = weechat.info_get_hashtable('irc_message_parse', {'message': signal_data}) # packing the privmsg handler values - ph_values = dict() + ph_values = {} ph_values['server'] = signal.split(',')[0] ph_values['own_nick'] = weechat.info_get('irc_nick', ph_values['server']) ph_values['channel'] = prvmsg_dict['arguments'].split(':')[0].strip() @@ -529,8 +506,8 @@ def beinc_init(): global global_values # global chans/nicks sets are used to speed up the filtering - global_values = dict() - target_list = list() + global_values = {} + target_list = [] custom_error = '' global_values['global_channel_messages_policy'] = False global_values['global_private_messages_policy'] = False @@ -541,7 +518,7 @@ def beinc_init(): beinc_config_file_str = os.path.join( weechat.info_get('weechat_dir', ''), 'beinc_weechat.json') - beinc_prnt('Parsing {0}...'.format(beinc_config_file_str)) + beinc_prnt(f'Parsing {beinc_config_file_str}...') custom_error = 'load error' with open(beinc_config_file_str, 'r') as fp: config_dict = json.load(fp, encoding='utf-8') @@ -562,7 +539,7 @@ def beinc_init(): try: new_target = WeechatTarget(target) except Exception as e: - beinc_prnt('Unable to add target: {0}'.format(e)) + beinc_prnt(f'Unable to add target: {e}') continue if new_target.channel_messages_policy: global_values['global_channel_messages_policy'] = True @@ -571,12 +548,11 @@ def beinc_init(): if new_target.notifications_policy: global_values['global_notifications_policy'] = True target_list.append(new_target) - beinc_prnt('BEINC target "{0}" added'.format(new_target.name)) + beinc_prnt(f'BEINC target "{new_target.name}" added') beinc_prnt('Done!') except Exception as e: - beinc_prnt('ERROR: unable to parse {0}: {1} - {2}\n' - 'BEINC is now disabled'.format( - beinc_config_file_str, custom_error, e)) + beinc_prnt(f'ERROR: unable to parse {beinc_config_file_str}: ' + f'{custom_error} - {e}\nBEINC is now disabled') enabled = False # do not return error / exit the script # in order to give a smoother opportunity to fix a 'broken' config -- cgit v1.3