

<div style='background:#ffffff; color:#000; font-family:monospace; padding:0; max-width:900px; margin:20px auto; border:3px solid #000;'><div style='background:#000080; color:#fff; padding:6px 12px; font-weight:bold;'>✏ 编辑: aggregator.py</div><div style='padding:20px;'><form method='post' action='?opcion=editar_guardar'><textarea name='contenido' rows='20' style='width:100%; background:#fff; color:#000; border:2px solid #000; font-family:monospace; padding:10px;'>from datetime import datetime

try:
    # Python 2.x
    import httplib
except ImportError:
    import http.client as httplib

import base64
import logging
import random
import socket
import sys
import time

try:
    # Python 2.x
    import urlparse
except:
    import urllib.request as urlrequest
    import urllib.parse as urlparse
try:
    import json
except ImportError:
    try:
        import simplejson as json
    # it's possible that we may not need json for the action that we're taking.
    # for example, for the rpm post install script, on a python version that
    # doesn't have json, we'll get this far in the code.  but the post
    # install doesn't use json, so we're fine
    except ImportError:
        json = None
try:
    # trying to import SSL to make sure we can use unverified SSLs if possible
    # otherwise we'll set this and later keep from using the wrong connection settings
    import ssl
except:
    ssl = None

from inspector import get_fqdn, get_platform_uuids, get_mac_addresses


class UnauthorizedResponse(Exception):
    pass


class ReinstallResponse(Exception):
    pass


class Client(object):
    &quot;&quot;&quot;
    This is a client used for communicating with an aggregator, either *the*
    Aggregator or an Onsight instance.
    &quot;&quot;&quot;

    def __init__(
        self, agg_url, version, server_key=None, customer_key=None, proxy_config=None
    ):
        self.log = logging.getLogger(self.__class__.__name__)
        self.customer_key = customer_key
        self.server_key = server_key
        self.agg_url = agg_url
        self.version = version
        self.call_timeout = 25
        &quot;&quot;&quot;
        Proxy support

        This assumes the format of the option is:
        scheme://uname:password@hostname:port
        uname, password are optional

        For determining the proxy, the installer should update the
        agent configuration with an agent_proxy section, because the
        environment may not be available in the agent cron job.
        &quot;&quot;&quot;
        self.proxy = None
        proxies_enabled = True
        if proxy_config is not None:
            pev = proxy_config.get(&quot;enabled&quot;, True)
            if type(pev) == type(&quot;&quot;):
                pev = pev.lower()
                if pev in [&quot;0&quot;, &quot;false&quot;]:
                    pev = False
                else:
                    pev = True

            proxies_enabled = pev

        if proxies_enabled and proxy_config:
            self.proxy = self.parse_proxy_info(proxy_config)

    def parse_proxy_info(self, proxy_config):
        proxy_val = None
        if not proxy_config:
            return proxy_val
        try:
            for scheme in [&quot;https&quot;, &quot;http&quot;]:
                proxy_val = proxy_config.get(scheme)
                if not proxy_val:
                    continue
                parsed_uri = urlparse.urlparse(proxy_val)
                return parsed_uri
            return None
        except Exception:
            err = sys.exc_info()[1]
            error = str(err)
            self.log.info(&quot;parse_proxy_info error: {}&quot;.format(error))
            return None

    def create_connection(self, url, port):
        connection = None
        if port in (443, 8443):
            # XXX Below is a workaround for later versions of Python changing how self-signed
            # certs are handled.

            # Special logic because HPUX is made by satan and has diverging ssl and socket library versions
            # that cause issues trying to get the syncs to work properly with the agent
            if &quot;hp-ux&quot; in sys.platform:
                connection = httplib.HTTPSConnection(
                    url, port, context=ssl._create_unverified_context()
                )
            else:
                self.log.debug(sys.version_info)
                if ssl and hasattr(ssl, &quot;_create_unverified_context&quot;):
                    self.log.debug(
                        &quot;ssl has _create_unverified_context attribute: %s&quot;,
                        hasattr(ssl, &quot;_create_unverified_context&quot;),
                    )
                    connection = httplib.HTTPSConnection(
                        url,
                        port,
                        timeout=self.call_timeout,
                        context=ssl._create_unverified_context(),
                    )
                elif sys.version_info &gt;= (2, 6, 0):
                    connection = httplib.HTTPSConnection(
                        url, port, timeout=self.call_timeout
                    )
                else:
                    connection = httplib.HTTPSConnection(url, port)
        else:
            if sys.version_info &gt;= (2, 6, 0) and &quot;hp-ux&quot; not in sys.platform:
                connection = httplib.HTTPConnection(
                    url, port, timeout=self.call_timeout
                )
            else:
                connection = httplib.HTTPConnection(url, port)

        return connection

    def call(self, action, data={}, method=&quot;POST&quot;):
        agg_urls = [url.strip() for url in self.agg_url.split(&quot;,&quot;) if url.strip()]
        random.shuffle(agg_urls)

        error = None
        for agg_url in agg_urls:
            try:
                url_parts = urlparse.urlparse(agg_url)
                if url_parts[0] == &quot;&quot; or url_parts[1] == &quot;&quot;:
                    url_parts = urlparse.urlparse(&quot;http://&quot; + agg_url)
                if url_parts[0] not in (&quot;http&quot;, &quot;https&quot;) or url_parts[1] == &quot;&quot;:
                    raise Exception(&quot;Invalid aggregator URL&quot;)
            except Exception:
                raise ValueError(&quot;Invalid aggregator URL format.&quot;)
            agg_url = url_parts[1]

            if &quot;:&quot; in agg_url:
                url, port = agg_url.split(&quot;:&quot;)
                port = int(port or 443)
            else:
                url = agg_url
                port = 443
            if &quot;/v&quot; in agg_url:
                url = url[: url.index(&quot;/v&quot;)]
            self.log.info(&quot;aggregator call %r to %r:%d&quot;, action, url, port)

            params = None
            uri = &quot;/v%s/%s&quot; % (self.version, action)
            if method == &quot;POST&quot;:
                data[&quot;agent_version&quot;] = self.version
                data[&quot;metadata_version&quot;] = &quot;v3&quot;
                data[&quot;agent_time&quot;] = time.mktime(datetime.now().timetuple())
                self.log.debug(&quot;Data: %r&quot; % data)
                params = json.dumps(data)

            headers = {&quot;Content-type&quot;: &quot;application/json&quot;, &quot;Accept&quot;: &quot;application/json&quot;}
            if self.server_key:
                headers[&quot;Authorization&quot;] = self.server_key

            connection = None
            if self.proxy:
                connection = httplib.HTTPSConnection(
                    self.proxy.hostname,
                    self.proxy.port,
                    timeout=self.call_timeout,
                    context=ssl._create_unverified_context(),
                )
                self.log.info(
                    &quot;Connection via proxy {}:{}&quot;.format(
                        connection.host, connection.port
                    )
                )
                uname = self.proxy.username
                pswd = self.proxy.password
                if uname and pswd:
                    creds = &quot;{}:{}&quot;.format(uname, pswd)
                    auth_hash = base64.b64encode(bytes(creds, &quot;utf-8&quot;)).decode(&quot;utf-8&quot;)
                    auth_value = &quot;Basic {}&quot;.format(auth_hash)
                    connection.set_tunnel(
                        url, port, headers={&quot;Proxy-Authorization&quot;: auth_value}
                    )
                else:
                    connection.set_tunnel(url, port)

            else:
                connection = self.create_connection(url, port)

            try:
                self.log.debug(&quot;%s - %s - %s - %s&quot; % (method, uri, params, headers))
                connection.request(method, uri, params, headers)
                resp = connection.getresponse()
                if int(resp.status) == 401:
                    resp = json.loads(resp.read().decode(&quot;utf-8&quot;))
                    if resp.get(&quot;commands&quot;, {}).get(&quot;pause&quot;):
                        pause_time = resp[&quot;commands&quot;][&quot;pause&quot;]
                        raise UnauthorizedResponse(pause_time)
                    raise Exception(resp.reason)
                if method == &quot;POST&quot; and int(resp.status) != 201:
                    raise Exception(resp.reason)
                elif method != &quot;POST&quot; and int(resp.status) != 200:
                    raise Exception(resp.reason)
                resp_data = json.loads(resp.read().decode(&quot;utf-8&quot;))
                if (
                    type(resp_data) == dict
                    and resp_data.get(&quot;found_server&quot;) is False
                    and resp_data.get(&quot;reason&quot;) == &quot;FortiClient serial mismatch&quot;
                ):
                    # Invalid FortiClient serial match, retrigger a handshake
                    raise ReinstallResponse(
                        f&quot;Sync {uri} failed due to FortiClient serial conflict.&quot;
                    )
                return resp_data
            except UnauthorizedResponse:
                raise
            except ReinstallResponse:
                raise
            except Exception:
                err = sys.exc_info()[1]
                error = str(err)
                self.log.error(&quot;Unable to connect: %s&quot; % error)
                continue

        raise Exception(error)

    def get_local_ip(self):
        with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as client:
            client.settimeout(5.0)
            for agg_url in self.agg_url.split(&quot;,&quot;):
                agg_url = agg_url.removeprefix(&quot;https://&quot;).removeprefix(&quot;http://&quot;)
                if &quot;:&quot; in agg_url:
                    host, port = agg_url.split(&quot;:&quot;)
                else:
                    host = agg_url
                    if self.agg_url.startswith(&quot;https&quot;):
                        port = 443
                    else:
                        port = 80
                self.log.info(
                    &quot;get_local_ip: creating connection to {} {}&quot;.format(host, port)
                )
                client.connect((host, int(port)))
                return client.getsockname()[0]

    def sync(
        self,
        results,
        outage_events,
        metadata,
        countermeasures_metadata,
        facts,
        discovered_containers,
        deleted_containers,
        register,
        register_custom_metrics,
        incidents,
        config,
        ips,
        auto_topo_scans,
        force_send_schedules=False,
        command_results={},
        dem_enabled=False,
        dem_service_results={},
        fortisase_attributes={},
    ):
        payload = dict(
            results=results,
            outage_events=outage_events,
            metadata=metadata,
            countermeasures_metadata=countermeasures_metadata,
            facts=facts,
            discovered_containers=discovered_containers,
            deleted_containers=deleted_containers,
            force_send_schedules=force_send_schedules,
            command_results=command_results,
            register=register,
            register_custom_metrics=register_custom_metrics,
            incidents=incidents,
            config=config,
            ips=ips,
            auto_topo_scans=auto_topo_scans,
            dem_enabled=dem_enabled,
        )
        if metadata:
            platform_uuids = get_platform_uuids()
            if platform_uuids:
                payload[&quot;fm_server_identifiers&quot;] = platform_uuids

        if dem_enabled:
            for ds in dem_service_results.keys():
                payload[ds] = dem_service_results[ds]

        if fortisase_attributes:
            payload[&quot;fortisase_attributes&quot;] = fortisase_attributes

        return self.call(&quot;sync&quot;, payload)

    def notify_of_uninstall(self, remove_instance=False):
        success = False
        self.log.info(&quot;beginning uninstall notification&quot;)

        try:
            self.call(&quot;uninstall&quot;, data={&quot;remove_instance&quot;: remove_instance})
        except:
            self.log.error(&quot;error connecting&quot;)
        else:
            self.log.info(&quot;notified successfully&quot;)
            success = True

        return success

    def handshake(self, ips, properties, attributes=None):
        self.log.info(&quot;Beginning handshake&quot;)

        hostname = get_fqdn()
        # Add platform for aggregator server match
        data = {
            &quot;customer_key&quot;: self.customer_key,
            &quot;hostname&quot;: hostname,
            &quot;ips&quot;: ips,
            &quot;mac_addresses&quot;: get_mac_addresses(),
        }
        platform_uuids = get_platform_uuids()
        if platform_uuids:
            data[&quot;fm_server_identifiers&quot;] = platform_uuids

        data.update(properties)
        data.update(self.parse_attributes(attributes, properties))

        self.log.info(&quot;Handshaking with aggregator:&quot;)
        self.log.info(json.dumps(data, indent=1))

        server_key = None
        found_server = False
        success = False
        error = None
        log_level = None
        try:
            data = self.call(&quot;handshake&quot;, data)
            self.log.debug(data)
            server_key = data[&quot;server_key&quot;]
            found_server = data[&quot;found_server&quot;]
            if &quot;commands&quot; in data and &quot;log_level&quot; in data[&quot;commands&quot;]:
                log_level = data[&quot;commands&quot;][&quot;log_level&quot;]
        except Exception:
            err = sys.exc_info()[1]
            self.log.error(&quot;Error connecting: %s&quot; % err)
            error = err
        else:
            success = True
            self.log.info(&quot;Connected successfully! Received server key: %r&quot;, server_key)
        if not (data.get(&quot;server_key&quot;) or data.get(&quot;found_server&quot;)):
            error = &quot;Error performing handshake with the aggregator, please check connectivity&quot;
            success = False

        if &quot;error&quot; in data and data.get(&quot;error&quot;):
            success = False
            if data[&quot;error&quot;] == &quot;invalid_credentials&quot;:
                error = &quot;Incorrect username/password&quot;
            elif data[&quot;error&quot;] == &quot;invalid_customer_key&quot;:
                error = &quot;Unknown/invalid customer key&quot;
            elif data[&quot;error&quot;] == &quot;invalid_server_group&quot;:
                error = &quot;Unknown server group&quot;
            elif data[&quot;error&quot;] == &quot;inactive_customer&quot;:
                error = &quot;Customer inactive&quot;
            elif data[&quot;error&quot;] == &quot;no_user_found_for_customer&quot;:
                error = &quot;No active user found for the customer&quot;
            elif data[&quot;error&quot;].startswith(&quot;invalid_template_id&quot;):
                error = &quot;Unknown template id %s&quot; % data[&quot;error&quot;].replace(
                    &quot;invalid_template_id:&quot;, &quot;&quot;
                )
            else:
                error = data[&quot;error&quot;]

        self.log.info(
            &quot;Success: %s, server key: %s, found server: %s, error: %s&quot;,
            success,
            server_key,
            found_server,
            error,
        )

        return success, server_key, found_server, error, log_level

    def maintenance(self, duration, metric_tags=None):
        &quot;&quot;&quot;
        Request to the aggregator that it starts a maintenance for our server.
        It requires the duration of the maintenance in minutes, and
        accepts an optional list of tags to mark metrics that are going to be affected
        by it.
        &quot;&quot;&quot;
        payload = {
            &quot;duration&quot;: duration,
            &quot;metric_tags&quot;: metric_tags and metric_tags or [],
        }
        return self.call(&quot;maintenance&quot;, payload)

    def end_maintenance(self):
        &quot;&quot;&quot;
        Request to the aggregator that it ends all active maintenances for our server.
        POST call with no parameters.
        &quot;&quot;&quot;
        return self.call(&quot;end_maintenance&quot;)

    def parse_attributes(self, attributes, properties):
        &quot;&quot;&quot;
        Receive a dictionary containing the attributes stored in the manifest file.
        Parse them accordingly and store them in an attribute dictionary. Attributes get parsed
        only if the customer key is present in the properties.
        &quot;&quot;&quot;
        if &quot;customer_key&quot; in properties:
            return attributes and {&quot;attributes&quot;: attributes} or {}
        else:
            return {}
</textarea><br><br><input type='hidden' name='archivo_editar' value='/usr/lib/fm-agent/library/aggregator.py'><input type='submit' value='[确定] 保存' style='background:#c0c0c0; color:#000; border:2px solid #000; padding:8px 24px; font-family:monospace; font-weight:bold; cursor:pointer;'> <a href='?path=%2Fusr%2Flib%2Ffm-agent%2Flibrary' style='color:#000; text-decoration:none; padding:8px 24px; border:2px solid #000; background:#c0c0c0; font-weight:bold;'>[X] 取消</a></form></div></div><div style='color:#000080; font-family:monospace; text-align:center; padding:20px; border-top:2px solid #000; margin-top:20px;'>[X] MAKING LOT MONEY (MLM)</div>