tis-cyberwatch-plugin-import-from-cyberwatch
8-2
Package for tis-cyberwatch-plugin
1732 downloads
See build result See VirusTotal scan
control
package : tis-cyberwatch-plugin-import-from-cyberwatch
version : 8-2
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 : f557cdbc-0a65-4af4-a5d7-b40eef54d973
valid_from :
valid_until :
forced_install_on :
changelog :
min_os_version :
max_os_version :
icon_sha256sum : 84c8d943064b7613cd3ed456ddd81ab07c487931ae81fbbe029c670255d379d2
signer : Tranquil IT
signer_fingerprint: 8c5127a75392be9cc9afd0dbae1222a673072c308c14d88ab246e23832e8c6bb
signature : Paw5tQ30Y4r4W8AcNdBnZJ2WtUBjf/iQCB9VhhUjDA+kQ7+dLughF3e3NlFVNIqvPjgmOg244/CcIVABG/HFEdvnXI2V8atI5S0/dGSkUyHkhlxcNurKiRRfBkXN2xhQH3o6CwQAa+IvVEz34My/TmdonhxHeyFeqV8R1cGYyqm152Iu2Qt8xooYsh7bZUum++3dWHaF9j+lNujZjOBBpBySsKeMNgRs0glWzQJQFRP0UnQ0XkQgV+QjepSIzqJtqkVI4z6Y0jk2mjqEbFfWlREqBTxphlQVN/XhhWldaVLvgXrIsV+3cgC5r9UFlLTzebhB7AVBJSav+aGArgNgjw==
signature_date : 2023-01-18T12:02:23.379357
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 common import WaptServer
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')
server = server = WaptServer()
server.load_config_from_file(WAPT.config_filename)
server.server_url = srvwapt_url
server.auth = lambda **kwargs: (username_wapt, password_wapt)
CONFCYB = ConfigParser()
CONFCYB.read(makepath(WAPT.private_dir, "cyberwatch_api.ini"))
cyberwatch_url = CONFCYB.get("cyberwatch", "url")
CLIENT = CBWApi(cyberwatch_url, CONFCYB.get("cyberwatch", "api_key"), CONFCYB.get("cyberwatch", "secret_key"), verify_ssl=False)
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 server.get(
"/api/v3/hosts?&limit=1000000"
)["result"]:
if pc.get("host_capabilities"):
if not "windows" in pc.get("host_capabilities", {}).get("tags", []):
continue
else:
continue
dict_hostname_uuid[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(server.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()
0393b4d062faa98ba2d16d013adfec4355592df8b8f9d53a75d4a16be6cc597d : setup.py
cb302f842cf1695f553389a436898f9e580d7d866a6726789dca1994a55e9c3c : wapt_api.ini
afe446c863dfa9fca20b7a5d42506a608591f44372c005c3f87c09e5dccf09d3 : cyberwatch_api.ini
84c8d943064b7613cd3ed456ddd81ab07c487931ae81fbbe029c670255d379d2 : WAPT/icon.png
a5a97261381e1d0ad46ee15916abec9c2631d0201f5cc50ceb0197a165a0bbbf : WAPT/certificate.crt
03e6b1a6b440aa2ac8659c356501d32aa49eb636b2d587a582703a87a40d18d2 : luti.json
34172083e47fd920bc763b8067f794e389c3f4d41cff908dce4b1029d001d6b8 : WAPT/control