tis-cyberwatch-plugin-import-from-cyberwatch
10-6
Paquet pour tis-cyberwatch-plugin
1570 téléchargements
Voir le résultat de la construction Voir l'analyse de VirusTotal
![tis-cyberwatch-plugin-import-from-cyberwatch icon](/wapt/icons/84c8d943064b7613cd3ed456ddd81ab07c487931ae81fbbe029c670255d379d2.png)
control
package : tis-cyberwatch-plugin-import-from-cyberwatch
version : 10-6
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 : d772abfb-0e66-41a6-ac1f-0c53ee456410
valid_from :
valid_until :
forced_install_on :
changelog :
min_os_version :
max_os_version :
icon_sha256sum : 84c8d943064b7613cd3ed456ddd81ab07c487931ae81fbbe029c670255d379d2
signer : Tranquil IT
signer_fingerprint: 8c5127a75392be9cc9afd0dbae1222a673072c308c14d88ab246e23832e8c6bb
signature : jO/76ULx7Vh6WHOqYPLQyAj53hTq6axX7tC0hxv9NIOoH+DI8h6dISnWWGiJVixKTUu7VUO8AZRze/bX9L7Vdmutbk/5c3eKbvIrdkjaHWoGfr2/8ahSX3ksz3/tFyKpCUQD5uDRZlykxOhZouD/wVadoPkCjMQUpcMMO5v9GkzzpMzXxTcP4BJ3Z6MyXWLWWJxkd9oWSo9oKsrrnUNqHan7w2sPF0EnEu/kk/g/N3oe/aH+dWTDOdF5Pnq8ojLG2BEZBjuEqPcDaHiFtVeGA6usB85u6HhA0Ds9fcQMKQVmeab9ByQgjT3ymgO3jnbHc0QjMs64M1Cmolq90NEp4w==
signature_date : 2024-07-21T09:00:50.060287
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
Setup.py
# -*- coding: utf-8 -*-
from setuphelpers import *
from configparser import ConfigParser
from waptutils import get_verify_cert
import platform
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")
# certifi import with system_redirection but install run without system_redirection so copy cacert.pem
if platform.system() == 'Windows':
if isfile(r'C:\Windows\SysWOW64\config\systemprofile\AppData\Local\.certifi\cacert.pem'):
mkdirs(r'C:\WINDOWS\system32\config\systemprofile\AppData\Local\.certifi')
filecopyto(r'C:\Windows\SysWOW64\config\systemprofile\AppData\Local\.certifi\cacert.pem',r'C:\WINDOWS\system32\config\systemprofile\AppData\Local\.certifi\cacert.pem')
WAPT.waptserver.post('api/v3/login' ,data=json.dumps({'user': username_wapt, 'password': password_wapt}))
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 = CBWApi(CONFCYB.get("cyberwatch", "url"), CONFCYB.get("cyberwatch", "api_key"), CONFCYB.get("cyberwatch", "secret_key"), verify_ssl=verify_cert_cyb)
dict_hostname_uuid = {}
def list_to_dict(newkey=None, listconvert=None):
newdict = {}
for p in listconvert:
newdict[p[newkey]] = p
return newdict
for pc in WAPT.waptserver.get(
"api/v3/hosts?&limit=1000000"
)["result"]:
dict_hostname_uuid[str(pc["computer_name"]).lower()] = pc["uuid"]
list_error = []
waptdata = []
for entry in CLIENT.servers():
# for entry in CLIENT.assets() :
if str(entry.hostname).lower() in dict_hostname_uuid:
try:
####dictionary conversion for easy searching in console###########
resultpc = json.loads(CLIENT._request("GET", ["/api/v3/vulnerabilities/servers", str(entry.id)]).content.decode("utf-8"))
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)))
print(WAPT.waptserver.post("api/v3/update_hosts_audit_data", data=json.dumps(waptdata)))
if list_error:
print(" \n".join(list_error))
return "ERROR"
else:
return "OK"
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 python lib
import json
import time
import datetime
import base64
import hmac
from hashlib import sha256
from requests.auth import AuthBase
import logging
import functools
import requests
from collections import namedtuple
from urllib.parse import urlparse, urljoin
from urllib.parse import parse_qs
from requests.exceptions import ProxyError, SSLError, RetryError, InvalidHeader, MissingSchema
from urllib3.exceptions import NewConnectionError, MaxRetryError
ROUTE_SERVERS = "/api/v3/vulnerabilities/servers"
SIGNATURE_HEADER = "CyberWatch APIAuth-HMAC-SHA256"
SIGNATURE_HTTP_HEADER = "Authorization"
TIMESTAMP_HTTP_HEADER = "Date"
SIGNATURE_DELIM = ":"
CONTENT_TYPE_HEADER = "Content-Type"
JSON_CONTENT_TYPE = "application/json"
class CBWApi: # pylint: disable=R0904
"""Class used to communicate with the CBW API"""
def __init__(
self,
api_url=None,
api_key=None,
secret_key=None,
verify_ssl=False,
):
self.verify_ssl = verify_ssl
self.logger = logging.getLogger(self.__class__.__name__)
try:
self.api_url = api_url
self.api_key = api_key
self.secret_key = secret_key
except KeyError as error:
error("CBWApi is missing one of its argument. You can use the environnement variable %s.")
def _build_route(self, params):
return functools.reduce(lambda base_url, arg: urljoin(f"{base_url}/", arg), [self.api_url] + params)
def _cbw_parser(self, response):
"""Parse the response text of an API request"""
try:
result = json.loads(response.text, object_hook=lambda d: namedtuple("cbw_object", d.keys())(*d.values()))
return result
except TypeError:
self.logger.error("An error occurred while parsing response")
sys.exit(-1)
def _request(self, verb, payloads, body_params=None, params=None):
route = self._build_route(payloads)
if body_params is not None:
body_params = json.dumps(body_params)
try:
return requests.request(verb, route, data=body_params, params=params, auth=CBWAuth(self.api_key, self.secret_key), verify=self.verify_ssl)
except (ConnectionError, ProxyError, SSLError, NewConnectionError, RetryError, InvalidHeader, MaxRetryError):
self.logger.exception("An error occurred when requesting {}".format(route))
sys.exit(-1)
except MissingSchema:
self.logger.error("An error occurred, please check your API_URL.")
sys.exit(-1)
def _get_pages(self, verb, route, params):
"""Get one or more pages for a method using api v3 pagination"""
response_list = []
if params is None:
params = {}
params["per_page"] = 100
if "per_page" not in params:
params["per_page"] = 100
if "page" in params:
response = self._request(verb, route, params)
if not response.ok:
logging.error("Error::{}".format(response.text))
return None
return self._cbw_parser(response)
response = self._request(verb, route, params)
if not response.ok:
logging.error("Error::{}".format(response.text))
return None
response_list.extend(self._cbw_parser(response))
while "next" in response.links:
next_url = urlparse(response.links["next"]["url"])
params["page"] = parse_qs(next_url.query)["page"][0]
response = self._request(verb, route, params)
response_list.extend(self._cbw_parser(response))
return response_list
def servers(self, params=None):
"""GET request to /api/v3/servers to get all servers"""
response = self._get_pages("GET", [ROUTE_SERVERS], params)
return response
class CBWAuth(AuthBase):
"""Used to make the authentication for the API requests"""
def __init__(self, api_key, secret_key):
self.api_key = api_key
self.secret_key = secret_key
self.raw_data = ""
self.hash_data = ""
self.type_data = ""
def __call__(self, request):
self._encode(request)
return request
def _encode(self, request):
timestamp = datetime.datetime.utcnow().strftime("%a, %d %b %Y %H:%M:%S GMT")
# get data
self.raw_data = request.body if request.body else ""
self.type_data = JSON_CONTENT_TYPE if self.raw_data else ""
# add headers
if self.raw_data:
self._add_content_type(request)
request.headers[TIMESTAMP_HTTP_HEADER] = timestamp
self._add_signature(request, timestamp)
def _add_signature(self, request, timestamp):
method = request.method
path = request.path_url
signature = self._sign(method, timestamp, path).decode("utf-8")
request.headers[SIGNATURE_HTTP_HEADER] = "{0} {1}:{2}".format(SIGNATURE_HEADER, self.api_key, signature)
def _add_content_type(self, request):
request.headers[CONTENT_TYPE_HEADER] = self.type_data
def _sign(self, method, timestamp, path):
# Build the message to sign
message = ",".join([method, self.type_data, self.hash_data, path, timestamp])
# Create the signature
digest = hmac.new(bytes(self.secret_key, "utf8"), bytes(message, "utf8"), sha256).digest()
return base64.b64encode(digest).strip()
b0e0c4f39519e4699744ea100679bc1f150514124e6b9f1979731ab07fd46d68 : setup.py
cb302f842cf1695f553389a436898f9e580d7d866a6726789dca1994a55e9c3c : wapt_api.ini
e5cb0662e2b27d07eec7a643f4cdaffef8d6dcf15164d9d121826cb12f7623f1 : cyberwatch_api.ini
84c8d943064b7613cd3ed456ddd81ab07c487931ae81fbbe029c670255d379d2 : WAPT/icon.png
a5a97261381e1d0ad46ee15916abec9c2631d0201f5cc50ceb0197a165a0bbbf : WAPT/certificate.crt
27720aaedb3cf616933e29e1c2e48ccda4bf697bc320b8b1035d65eba5540c12 : luti.json
80a99d6468013de9f7445c0f40a6e4539ae2bd31b5fde69e9786f0daf3fef83c : .gitignore
725618d5fe673b08d7cca8577a33d1494312b0520f14b1b8829abcd92baf7851 : WAPT/control