support zabbix 6.4

[vasya3793]

1. for information
   i found that the pyzabbix project is like your py-zabbix project
   creates a pyzabbix library and works with it.
   when installing both, api.py is overwritten

2. I propose to modify to support zabbix 6.4

• remove the login from the init function, make it in the login procedure
  I attach the source:

-- encoding: utf-8 --

Copyright © 2014 Alexey Dubkov

This file is part of py-zabbix.

Py-zabbix 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.

Py-zabbix 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 py-zabbix. If not, see http://www.gnu.org/licenses/.

import json
import logging
import os
import ssl
import sys
import base64
from packaging.version import Version

For Python 2 and 3 compatibility

try:
import urllib2
except ImportError:
# Since Python 3, urllib2.Request and urlopen were moved to
# the urllib.request.
import urllib.request as urllib2

from .version import version
from .logger import NullHandler, HideSensitiveFilter, HideSensitiveService

null_handler = NullHandler()
logger = logging.getLogger(name)
logger.addHandler(null_handler)
logger.addFilter(HideSensitiveFilter())

class ZabbixAPIException(Exception):
"""ZabbixAPI exception class.

:code list:
:32602: Invalid params (eg already exists)
:32500: No permissions
"""
def __init__(self, *args):
    super(Exception, self).__init__(*args)
    if len(args) == 1 and isinstance(args[0], dict):
        self.error = args[0]
        self.error['json'] = HideSensitiveService.hide_sensitive(self.error['json'])
        self.message = self.error['message']
        self.code = self.error['code']
        self.data = self.error['data']
        self.json = self.error['json']

class ZabbixAPIObjectClass(object):
"""ZabbixAPI Object class.

:type group: str
:param group: Zabbix API method group name.
    Example: `apiinfo.version` method it will be `apiinfo`.

:type parent: :class:`zabbix.api.ZabbixAPI` object
:param parent: ZabbixAPI object to use as parent.
"""

def __init__(self, group, parent):
    self.group = group
    self.parent = parent

def __getattr__(self, name):
    """Dynamically create a method.

    :type name: str
    :param name: Zabbix API method name.
        Example: `apiinfo.version` method it will be `version`.
    """

    def fn(*args, **kwargs):
        if args and kwargs:
            raise TypeError("Found both args and kwargs")

        method = '{0}.{1}'.format(self.group, name)
        logger.debug("Call %s method", method)

        return self.parent.do_request(
            method,
            args or kwargs
        )['result']

    return fn

def ssl_context_compat(func):
def inner(req):
# We should explicitly disable cert verification to support
# self-signed certs with urllib2 since Python 2.7.9 and 3.4.3

    default_version = (2, 7, 9)
    version = {
        2: default_version,
        3: (3, 4, 3),
    }

    python_version = sys.version_info[0]
    minimum_version = version.get(python_version, default_version)

    if sys.version_info[0:3] >= minimum_version:
        # Create default context to skip SSL cert verification.
        ctx = ssl.create_default_context()
        ctx.check_hostname = False
        ctx.verify_mode = ssl.CERT_NONE
        res = func(req, context=ctx)
    else:
        res = func(req)

    return res

return inner

@ssl_context_compat
def urlopen(*args, **kwargs):
return urllib2.urlopen(*args, **kwargs)

class ZabbixAPI(object):
"""ZabbixAPI class, implement interface to zabbix api.

:type url: str
:param url: URL to zabbix api. Default: `ZABBIX_URL` or
    `https://localhost/zabbix`

:type use_authenticate: bool
:param use_authenticate: Use `user.authenticate` method if `True` else
    `user.login`.

:type use_basic_auth: bool
:param use_basic_auth: Using basic auth if `True`

:type user: str
:param user: Zabbix user name. Default: `ZABBIX_USER` or `'Admin'`.

:type password: str
:param password: Zabbix user password. Default `ZABBIX_PASSWORD` or
    `zabbix`.

>>> from pyzabbix import ZabbixAPI
>>> z = ZabbixAPI('https://zabbix.server', user='Admin', password='zabbix')
>>> # Get API Version
>>> z.api_info.version()
>>> u'2.2.1'
>>> # or
>>> z.do_request('apiinfo.version')
>>> {u'jsonrpc': u'2.0', u'result': u'2.2.1', u'id': u'1'}
>>> # Get all disabled hosts
>>> z.host.get(status=1)
>>> # or
>>> z.do_request('host.getobjects', {'status': 1})
"""

def __init__(self, url=None, use_authenticate=False, use_basic_auth=False, user=None,
             password=None):

    url = url or os.environ.get('ZABBIX_URL') or 'https://localhost/zabbix'

user = user or os.environ.get('ZABBIX_USER') or 'Admin'

password = password or os.environ.get('ZABBIX_PASSWORD') or 'zabbix'

    self.use_authenticate = use_authenticate
    self.use_basic_auth = use_basic_auth
    self.auth = None
    self.url = url + '/api_jsonrpc.php'
    self.base64_cred = self.cred_to_base64(user, password) if self.use_basic_auth else None
    if user != None and password != None:
        self._login(user, password)
    logger.debug("JSON-PRC Server: %s", self.url)

def login(self, user='', password=''):
    self._login(user, password)

def __getattr__(self, name):
    """Dynamically create an object class (ie: host).

    :type name: str
    :param name: Zabbix API method group name.
        Example: `apiinfo.version` method it will be `apiinfo`.
    """

    return ZabbixAPIObjectClass(name, self)

def _login(self, user='', password=''):
    """Do login to zabbix server.

    :type user: str
    :param user: Zabbix user

    :type password: str
    :param password: Zabbix user password
    """

    logger.debug("ZabbixAPI.login({0},{1})".format(user, HideSensitiveService.HIDEMASK))

    if self._detect_version:
        self.version = Version(self.api_version())
        logger.info(f"Zabbix API version is: {self.version}")

    self.auth = None

if self.use_authenticate:

self.auth = self.user.authenticate(user=user, password=password)

else:

self.auth = self.user.login(user=user, password=password)

    if self.use_authenticate:
        self.auth = self.user.authenticate(user=user, password=password)
    elif self.version and self.version >= Version("5.4.0"):
        self.auth = self.user.login(username=user, password=password)
    else:
        self.auth = self.user.login(user=user, password=password)



def _logout(self):
    """Do logout from zabbix server."""

    if self.auth:
        logger.debug("ZabbixAPI.logout()")

        if self.user.logout():
            self.auth = None

def __enter__(self):
    return self

def __exit__(self, *args):
    self._logout()

@staticmethod
def cred_to_base64(user, password):
    """Create header for basic authorization
    :type user: str
    :param user: Zabbix user

    :type password: str
    :param password: Zabbix user password
    :return: str
    """
    base64string = base64.b64encode('{}:{}'.format(user, password).encode())
    return base64string.decode()

def api_version(self):
    """Return version of server Zabbix API.

    :rtype: str
    :return: Version of server Zabbix API.
    """

    return self.apiinfo.version()

def do_request(self, method, params=None):
    """Make request to Zabbix API.

    :type method: str
    :param method: ZabbixAPI method, like: `apiinfo.version`.

    :type params: str
    :param params: ZabbixAPI method arguments.

    >>> from pyzabbix import ZabbixAPI
    >>> z = ZabbixAPI()
    >>> apiinfo = z.do_request('apiinfo.version')
    """

    request_json = {
        'jsonrpc': '2.0',
        'method': method,
        'params': params or {},
        'id': '1',
    }

    # apiinfo.version and user.login doesn't require auth token
    if self.auth and (method not in ('apiinfo.version', 'user.login')):
        request_json['auth'] = self.auth

    logger.debug(
        'urllib2.Request({0}, {1})'.format(
            self.url,
            json.dumps(request_json)))

    data = json.dumps(request_json)
    if not isinstance(data, bytes):
        data = data.encode("utf-8")

    req = urllib2.Request(self.url, data)
    req.get_method = lambda: 'POST'
    req.add_header('Content-Type', 'application/json-rpc')
    req.add_header('User-Agent', 'py-zabbix/{}'.format(__version__))

    if self.use_basic_auth:
        req.add_header("Authorization", "Basic {}".format(self.base64_cred))

    try:
        res = urlopen(req)
        res_str = res.read().decode('utf-8')
        res_json = json.loads(res_str)
    except ValueError as e:
        raise ZabbixAPIException("Unable to parse json: %s" % e.message)

    res_str = json.dumps(res_json, indent=4, separators=(',', ': '))
    logger.debug("Response Body: %s", res_str)

    if 'error' in res_json:
        err = res_json['error'].copy()
        err.update({'json': str(request_json)})
        raise ZabbixAPIException(err)

    return res_json

def get_id(self, item_type, item=None, with_id=False, hostid=None, **args):
    """Return id or ids of zabbix objects.

    :type item_type: str
    :param item_type: Type of zabbix object. (eg host, item etc.)

    :type item: str
    :param item: Name of zabbix object. If it is `None`, return list of
        all objects in the scope.

    :type with_id: bool
    :param with_id: Returned values will be in zabbix json `id` format.
        Examlpe: `{'itemid: 128}`

    :type name: bool
    :param name: Return name instead of id.

    :type hostid: int
    :param hostid: Filter objects by specific hostid.

    :type templateids: int
    :param tempateids: Filter objects which only belong to specific
        templates by template id.

    :type app_name: str
    :param app_name: Filter object which only belong to specific
        application.

    :rtype: int or list
    :return: Return single `id`, `name` or list of values.
    """

    result = None
    name = args.get('name', False)

    type_ = '{item_type}.get'.format(item_type=item_type)

    item_filter_name = {
        'mediatype': 'description',
        'trigger': 'description',
        'triggerprototype': 'description',
        'user': 'alias',
        'usermacro': 'macro',
    }

    item_id_name = {
        'discoveryrule': 'item',
        'graphprototype': 'graph',
        'hostgroup': 'group',
        'itemprototype': 'item',
        'map': 'selement',
        'triggerprototype': 'trigger',
        'usergroup': 'usrgrp',
        'usermacro': 'hostmacro',
    }

    filter_ = {
        'filter': {
            item_filter_name.get(item_type, 'name'): item,
        },
        'output': 'extend'}

    if hostid:
        filter_['filter'].update({'hostid': hostid})

    if args.get('templateids'):
        if item_type == 'usermacro':
            filter_['hostids'] = args['templateids']
        else:
            filter_['templateids'] = args['templateids']

    if args.get('app_name'):
        filter_['application'] = args['app_name']

    logger.debug(
        'do_request( "{type}", {filter} )'.format(
            type=type_,
            filter=filter_))
    response = self.do_request(type_, filter_)['result']

    if response:
        item_id_str = item_id_name.get(item_type, item_type)
        item_id = '{item}id'.format(item=item_id_str)
        result = []
        for obj in response:
            # Check if object not belong current template
            if args.get('templateids'):
                if (not obj.get('templateid') in ("0", None) or
                        not len(obj.get('templateids', [])) == 0):
                    continue

            if name:
                o = obj.get(item_filter_name.get(item_type, 'name'))
                result.append(o)
            elif with_id:
                result.append({item_id: int(obj.get(item_id))})
            else:
                result.append(int(obj.get(item_id)))

        list_types = (list, type(None))
        if not isinstance(item, list_types):
            result = result[0]

    return result