tis-cyberwatch-plugin-import-from-cyberwatch

package           : tis-cyberwatch-plugin-import-from-cyberwatch
version           : 15-0
architecture      : all
section           : base
priority          : optional
name              : 
categories        : 
maintainer        : sfonteneau
description       : Package for tis-cyberwatch-plugin
depends           : 
conflicts         : 
maturity          : PROD
locale            : all
target_os         : all
min_wapt_version  : 2.0
sources           : 
installed_size    : 
impacted_process  : 
description_fr    : Paquet pour tis-cyberwatch-plugin
description_pl    : Pakiet dla tis-cyberwatch-plugin
description_de    : Paket für tis-cyberwatch-plugin
description_es    : Paquete para tis-cyberwatch-plugin
description_pt    : Pacote para tis-cyberwatch-plugin
description_it    : Pacchetto per tis-cyberwatch-plugin
description_nl    : Pakket voor tis-cyberwatch-plugin
description_ru    : Пакет для tis-cyberwatch-plugin
audit_schedule    : 1h
editor            : 
keywords          : 
licence           : 
homepage          : 
package_uuid      : 3f497a6b-34d7-4ca5-9783-a19198ac2eb9
valid_from        : 
valid_until       : 
forced_install_on : 
changelog         : 
min_os_version    : 
max_os_version    : 
icon_sha256sum    : 84c8d943064b7613cd3ed456ddd81ab07c487931ae81fbbe029c670255d379d2
signer            : Tranquil IT
signer_fingerprint: 8c5127a75392be9cc9afd0dbae1222a673072c308c14d88ab246e23832e8c6bb
signature_date    : 2025-10-07T11:02:34.000000
signed_attributes : package,version,architecture,section,priority,name,categories,maintainer,description,depends,conflicts,maturity,locale,target_os,min_wapt_version,sources,installed_size,impacted_process,description_fr,description_pl,description_de,description_es,description_pt,description_it,description_nl,description_ru,audit_schedule,editor,keywords,licence,homepage,package_uuid,valid_from,valid_until,forced_install_on,changelog,min_os_version,max_os_version,icon_sha256sum,signer,signer_fingerprint,signature_date,signed_attributes
signature         : hNgFnNdmjLKLoyFU/hyzRbGlX9BdAuouqfTZuYmok0geQZCP2Ge9Hb0XoYcUVLXAIKZEL6Al7frJi1HxofuIb7R4UT3LAhoaSzCkw136ZwnSImwDr6sLs70iOSpCSI2utKG8gRGPhq1ev1UI51rPhzdA45+Djj/u0vgoY1g/tWsmZjcnBz5ZmvZ+ACTd/tAz+1H2wML51k+r3j+fkhZnEEb5jsRvrOYQW/XfK1Q5big67wOnAr4QZO88eEydYhCmqkvwfFbyA2od01qw8/il/o6RPsuLYS4HQgcPF03cNtAWXpLarIox9s2demfrE8y2UbJGW5cWva0GRkcTV5XrHg==

# -*- coding: utf-8 -*-
from setuphelpers import *
from configparser import ConfigParser
from waptutils import get_verify_cert
from common import get_requests_client_cert_session
import platform
import waptlicences
import json
import concurrent.futures
import datetime

simultaneous_maximum_send=10

def audit():
    CONFWAPT = ConfigParser()
    CONFWAPT.read(makepath(WAPT.private_dir, "wapt_api.ini"))
    username_wapt = CONFWAPT.get("wapt", "username")
    password_wapt = CONFWAPT.get("wapt", "password")
    srvwapt_url = CONFWAPT.get("wapt", "url")


    t = waptlicences.waptserver_login(WAPT.config_filename,username_wapt,password_wapt)
    if not 'session' in t['session_cookies']:
        session_cookies = [u for u in t['session_cookies'] if u['Domain'] == WAPT.waptserver.server_url.split('://')[-1]][0]
    else:
        session_cookies = t['session_cookies']['session']
        session_cookies['Name'] = 'session'

    sessionwapt = get_requests_client_cert_session(WAPT.waptserver.server_url,cert=(t['client_certificate'],t['client_private_key'],t['client_private_key_password']),verify=WAPT.waptserver.verify_cert)
    sessionwapt.cookies.set(session_cookies['Name'], session_cookies['Value'], domain=session_cookies['Domain'])
    sessionwapt.verify = WAPT.waptserver.verify_cert

    CONFCYB = ConfigParser()
    CONFCYB.read(makepath(WAPT.private_dir, "cyberwatch_api.ini"))
    cyberwatch_url = CONFCYB.get("cyberwatch", "url")

    if CONFCYB.has_option("cyberwatch", 'verify_cert'):
        verify_cert_cyb = get_verify_cert(CONFCYB.get("cyberwatch", 'verify_cert'))
    else:
        verify_cert_cyb = True

    CLIENT = Cyberwatch_Pyhelper()
    CLIENT.api_url = CONFCYB.get("cyberwatch", "url")
    CLIENT.api_key = CONFCYB.get("cyberwatch", "api_key")
    CLIENT.api_secret = CONFCYB.get("cyberwatch", "secret_key")
    CLIENT.verify_ssl = verify_cert_cyb

    dict_hostname_uuid = {}

    for pc in json.loads(sessionwapt.get("%s/api/v3/hosts?&limit=1000000" % WAPT.waptserver.server_url).content)["result"]:
        dict_hostname_uuid[str(pc["computer_name"]).lower()] = pc["uuid"]

    list_error = []


    list_run = []

    for hostname, uuid in dict_hostname_uuid.items():
        list_run.append({
            "sessionwapt": sessionwapt,
            "CLIENT": CLIENT,
            "cyberwatch_url": cyberwatch_url,
            "dict_hostname_uuid": {hostname: uuid},
            "WAPT": WAPT,
            "hostname":hostname,
            "list_error":list_error
        })

    results = []
    with concurrent.futures.ThreadPoolExecutor(simultaneous_maximum_send) as executor:
        futures = {executor.submit(import_from_cyberwatch, **g): g for g in list_run}

        for future in concurrent.futures.as_completed(futures):
            results.append(future.result())


    waptlicences.waptserver_logout(WAPT.config_filename)

    if list_error:
        print(" \n".join(list_error))
        return "ERROR"
    else:
        return "OK"

def import_from_cyberwatch(sessionwapt=None,CLIENT=None,cyberwatch_url=None,dict_hostname_uuid=None,WAPT=None,hostname=None,list_error=None):

        waptdata = []
        resultcyb = next(CLIENT.request(method="GET",endpoint="/api/v3/vulnerabilities/servers?hostname=%s" % hostname)).json()

        if not resultcyb:
            return
        entry=resultcyb[0]

        # for entry in CLIENT.assets() :
        if str(entry['hostname']).lower() in dict_hostname_uuid:
            try:
                ####dictionary conversion for easy searching in console###########
                resultpc = next(CLIENT.request(method="GET",endpoint="/api/v3/vulnerabilities/servers/%s" % str(entry['id']))).json()

                list_patch = {}
                for patch in resultpc["updates"]:
                    if patch.get("target", {}):
                        if patch.get("cve_announcements", []):
                            list_patch[patch.get("target", {}).get("product", "")]=None
                resultpc['cyberwatch_url'] = cyberwatch_url
                resultpc["list_patch"] = sorted(list(list_patch))

                resultpc["updates"] = list_to_dict(newkey="id", listconvert=resultpc["updates"])
                resultpc["security_issues"] = list_to_dict(newkey="id", listconvert=resultpc["security_issues"])
                resultpc["groups"] = list_to_dict(newkey="name", listconvert=resultpc["groups"])
                resultpc["cve_announcements"] = list_to_dict(
                    newkey="cve_code", listconvert=[cve for cve in resultpc["cve_announcements"] if cve["active"]]
                )
                resultpc["compliance_repositories"] = list_to_dict(newkey="id", listconvert=resultpc["compliance_repositories"])

                ##################################################################

                waptdata.append(
                    {
                        "host_id": dict_hostname_uuid[str(entry['hostname']).lower()],
                        "value_id": int(time.monotonic() * 1000),
                        "value_date": datetime2isodate(datetime.datetime.utcnow()),
                        "value_section": "cyberwatch",
                        "value_key": "cyberwatch",
                        "value": resultpc,
                        "expiration_date": datetime2isodate(datetime.datetime.utcnow() + datetime.timedelta(minutes=1440)),
                    }
                )
                print("Get %s from cyberwatch" % str(entry['hostname']).lower())
                time.sleep(0.0000001)

            except Exception as e:
                list_error.append("Error %s %s" % (str(entry['hostname']).lower(), str(e)))

            sessionwapt.post("%s/api/v3/update_hosts_audit_data" % WAPT.waptserver.server_url, data=json.dumps(waptdata))



def install():
    if not isfile(makepath(WAPT.private_dir, "cyberwatch_api.ini")):
        filecopyto("cyberwatch_api.ini", makepath(WAPT.private_dir, "cyberwatch_api.ini"))
    if not isfile(makepath(WAPT.private_dir, "wapt_api.ini")):
        filecopyto("wapt_api.ini", makepath(WAPT.private_dir, "wapt_api.ini"))


"""Cyberwatch API Python Helper"""
import os
import json
from configparser import ConfigParser
from typing import Generator
import requests

class Cyberwatch_Pyhelper():

    def __init__(self):
        self.api_url = None
        self.api_key = None
        self.api_secret = None

    def clear_endpoint(f):
        """
        Decorator that takes the endpoint that was given by the API user,
        and replaces the {parameter} by the one that was given inside the params or body params dict
        """
        def wrapper(*args, **kwargs):
            endpoint = kwargs.get("endpoint")
            parameters = [parameter.split("}")[0] for parameter in endpoint.split("{")[1:]]

            for parameter in parameters:
                parameter_value = (kwargs.get("params",{}).get(parameter) or kwargs.get("body_params",{}).get(parameter))
                endpoint = endpoint.replace("{" + parameter + "}", str(parameter_value))
                # Deleting the 'parameter' from the kwargs arguments if it exists
                [kwargs[key].pop(parameter, "") for key in kwargs if isinstance(kwargs[key], dict)]

            kwargs.update({"endpoint": endpoint})
            return f(*args, **kwargs)
        return wrapper

    def __basic_auth(self) -> requests.auth.HTTPBasicAuth:
        """
        Private method returning a BasicAuth
        """
        return requests.auth.HTTPBasicAuth(self.api_key, self.api_secret)

    @clear_endpoint
    def request(self, **kwargs) -> Generator[requests.models.Response, None, None]:
        """
        Only accessible method, handles every step of the API call
        """
        if not isinstance(kwargs.get("method"), str):
            raise Exception("The type of endpoint parameter should be str")
        if kwargs.get("timeout") and not isinstance(kwargs.get("timeout"), int):
            raise Exception("The type of timeout parameter should be int")

        method = str(kwargs.get("method")).upper()
        timeout = kwargs.get("timeout") or 10
        params = kwargs.get("params") or {}
        body_params = json.dumps(kwargs.get("body_params")) if kwargs.get("body_params") else {}
        headers = {'Content-type': 'application/json'}
        url = self.api_url + kwargs.get("endpoint")

        while url:
            response = requests.request(
                method=method,
                url=url,
                headers=headers,
                auth=self.__basic_auth(),
                params=params,
                data=body_params,
                timeout=timeout,
                verify=self.verify_ssl
            )
            url = response.links["next"]["url"] if "next" in response.links else None
            yield response

def list_to_dict(newkey=None, listconvert=None):
    newdict = {}
    for p in listconvert:
        newdict[p[newkey]] = p
    return newdict

80a99d6468013de9f7445c0f40a6e4539ae2bd31b5fde69e9786f0daf3fef83c : .gitignore
38d056ab130f7bf7c481c12636a4e9959de36561d3dfcbe54c6e3571bc0c1dc3 : WAPT/certificate.crt
badaeb89036ab7bea43989e60f3937636666ef8fb619b71fbb35a51fe702b1b3 : WAPT/control
84c8d943064b7613cd3ed456ddd81ab07c487931ae81fbbe029c670255d379d2 : WAPT/icon.png
e5cb0662e2b27d07eec7a643f4cdaffef8d6dcf15164d9d121826cb12f7623f1 : cyberwatch_api.ini
528f79605021429b4628ca3f3375ed0e9afa02e258e382b8703c2ec22db458a8 : luti.json
7a9e4e6eefb85d3de6ca0cf530f5047030881e5dafa53e32d78ba2ae8287858c : setup.py
cb302f842cf1695f553389a436898f9e580d7d866a6726789dca1994a55e9c3c : wapt_api.ini