adminmpmcz/patchcenter
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 2 Wiki Activity

fix(qualys): API 2.0 confirmee (3.0 OTHER_ERROR car endpoint inexistant, 5.0 INVALID_API_VERSION) + body XML + mapping field user-friendly vers nom Qualys (hostname->name, fqdn->dnsHostName, ip->address) - cause racine 401 = mot de passe API expire

Browse Source
This commit is contained in:
Pierre & Lumière 2026-05-05 16:56:56 +02:00
parent cadef89c50
commit c77b4b22eb
3 changed files with 34 additions and 17 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

43
app/services/qualys_service.py
View File

@ -32,8 +32,25 @@ def parse_xml(txt, tag):
return re.findall(f"<{tag}>([^<]*)</{tag}>", txt) return re.findall(f"<{tag}>([^<]*)</{tag}>", txt)
# Mapping : noms "user-friendly" → noms exacts attendus par l'API Qualys 2.0
# cf https://docs.qualys.com/en/am/api — search/am/hostasset criteria fields
QUALYS_FIELD_MAP = {
"hostname": "name",
"host": "name",
"fqdn": "dnsHostName",
"dns": "dnsHostName",
"ip": "address",
"ip_address": "address",
"id": "id",
"asset_id": "id",
"netbios": "netbiosName",
}
def search_assets_api(db, query, field="name", operator="CONTAINS", force_refresh=False): def search_assets_api(db, query, field="name", operator="CONTAINS", force_refresh=False):
"""Recherche des assets via l'API Qualys — cache 10 min""" """Recherche des assets via l'API Qualys — cache 10 min."""
# Traduit le field front vers le field Qualys (sinon "INVALID_FIELD")
field = QUALYS_FIELD_MAP.get((field or "").lower(), field)
cache_key = f"qualys:search:{field}:{query}" cache_key = f"qualys:search:{field}:{query}"
if not force_refresh: if not force_refresh:
@ -60,7 +77,7 @@ def search_assets_api(db, query, field="name", operator="CONTAINS", force_refres
"</ServiceRequest>" "</ServiceRequest>"
) )
r = requests.post( r = requests.post(
f"{qualys_url}/qps/rest/3.0/search/am/hostasset", f"{qualys_url}/qps/rest/2.0/search/am/hostasset",
data=xml_body, data=xml_body,
auth=(qualys_user, qualys_pass), auth=(qualys_user, qualys_pass),
verify=False, timeout=60, proxies=proxies, verify=False, timeout=60, proxies=proxies,
@ -98,7 +115,7 @@ def get_all_tags_api(db):
try: try:
r = requests.post( r = requests.post(
f"{qualys_url}/qps/rest/3.0/search/am/tag", f"{qualys_url}/qps/rest/2.0/search/am/tag",
json={"ServiceRequest": {"preferences": {"limitResults": 1000}}}, json={"ServiceRequest": {"preferences": {"limitResults": 1000}}},
auth=(qualys_user, qualys_pass), auth=(qualys_user, qualys_pass),
verify=False, timeout=60, proxies=proxies, verify=False, timeout=60, proxies=proxies,
@ -169,7 +186,7 @@ def create_tag_api(db, tag_name):
proxies = {"https": qualys_proxy, "http": qualys_proxy} if qualys_proxy else None proxies = {"https": qualys_proxy, "http": qualys_proxy} if qualys_proxy else None
try: try:
r = requests.post( r = requests.post(
f"{qualys_url}/qps/rest/3.0/create/am/tag", f"{qualys_url}/qps/rest/2.0/create/am/tag",
json={"ServiceRequest": {"data": {"Tag": {"name": tag_name}}}}, json={"ServiceRequest": {"data": {"Tag": {"name": tag_name}}}},
auth=(qualys_user, qualys_pass), verify=False, timeout=30, proxies=proxies, auth=(qualys_user, qualys_pass), verify=False, timeout=30, proxies=proxies,
headers={"Content-Type": "application/json"}) headers={"Content-Type": "application/json"})
@ -195,7 +212,7 @@ def delete_tag_api(db, qualys_tag_id):
proxies = {"https": qualys_proxy, "http": qualys_proxy} if qualys_proxy else None proxies = {"https": qualys_proxy, "http": qualys_proxy} if qualys_proxy else None
try: try:
r = requests.post( r = requests.post(
f"{qualys_url}/qps/rest/3.0/delete/am/tag/{qualys_tag_id}", f"{qualys_url}/qps/rest/2.0/delete/am/tag/{qualys_tag_id}",
auth=(qualys_user, qualys_pass), verify=False, timeout=30, proxies=proxies, auth=(qualys_user, qualys_pass), verify=False, timeout=30, proxies=proxies,
headers={"Content-Type": "application/json"}) headers={"Content-Type": "application/json"})
if r.status_code == 200 and "SUCCESS" in r.text: if r.status_code == 200 and "SUCCESS" in r.text:
@ -216,7 +233,7 @@ def add_tag_to_asset_api(db, asset_id, tag_id):
proxies = {"https": qualys_proxy, "http": qualys_proxy} if qualys_proxy else None proxies = {"https": qualys_proxy, "http": qualys_proxy} if qualys_proxy else None
try: try:
r = requests.post( r = requests.post(
f"{qualys_url}/qps/rest/3.0/update/am/hostasset/{asset_id}", f"{qualys_url}/qps/rest/2.0/update/am/hostasset/{asset_id}",
json={"ServiceRequest": {"data": {"HostAsset": {"tags": {"add": {"TagSimple": {"id": tag_id}}}}}}}, json={"ServiceRequest": {"data": {"HostAsset": {"tags": {"add": {"TagSimple": {"id": tag_id}}}}}}},
auth=(qualys_user, qualys_pass), verify=False, timeout=30, proxies=proxies, auth=(qualys_user, qualys_pass), verify=False, timeout=30, proxies=proxies,
headers={"Content-Type": "application/json"}) headers={"Content-Type": "application/json"})
@ -240,7 +257,7 @@ def remove_tag_from_asset_api(db, asset_id, tag_id):
proxies = {"https": qualys_proxy, "http": qualys_proxy} if qualys_proxy else None proxies = {"https": qualys_proxy, "http": qualys_proxy} if qualys_proxy else None
try: try:
r = requests.post( r = requests.post(
f"{qualys_url}/qps/rest/3.0/update/am/hostasset/{asset_id}", f"{qualys_url}/qps/rest/2.0/update/am/hostasset/{asset_id}",
json={"ServiceRequest": {"data": {"HostAsset": {"tags": {"remove": {"TagSimple": {"id": tag_id}}}}}}}, json={"ServiceRequest": {"data": {"HostAsset": {"tags": {"remove": {"TagSimple": {"id": tag_id}}}}}}},
auth=(qualys_user, qualys_pass), verify=False, timeout=30, proxies=proxies, auth=(qualys_user, qualys_pass), verify=False, timeout=30, proxies=proxies,
headers={"Content-Type": "application/json"}) headers={"Content-Type": "application/json"})
@ -300,7 +317,7 @@ def sync_server_qualys(db, server_id):
# Recuperer l'asset complet avec tags # Recuperer l'asset complet avec tags
try: try:
r = requests.post( r = requests.post(
f"{qualys_url}/qps/rest/3.0/search/am/hostasset", f"{qualys_url}/qps/rest/2.0/search/am/hostasset",
json={"ServiceRequest": { json={"ServiceRequest": {
"filters": {"Criteria": [ "filters": {"Criteria": [
{"field": "id", "operator": "EQUALS", "value": str(qid)} {"field": "id", "operator": "EQUALS", "value": str(qid)}
@ -388,7 +405,7 @@ def _find_asset_by_hostname(qualys_url, qualys_user, qualys_pass, hostname, prox
"""Cherche un asset Qualys par hostname""" """Cherche un asset Qualys par hostname"""
try: try:
r = requests.post( r = requests.post(
f"{qualys_url}/qps/rest/3.0/search/am/hostasset", f"{qualys_url}/qps/rest/2.0/search/am/hostasset",
json={"ServiceRequest": { json={"ServiceRequest": {
"preferences": {"limitResults": 5}, "preferences": {"limitResults": 5},
"filters": {"Criteria": [ "filters": {"Criteria": [
@ -688,7 +705,7 @@ def _refresh_all_agents_impl(db, mode="diff"):
for attempt in range(3): for attempt in range(3):
try: try:
r = sess.post( r = sess.post(
f"{qualys_url}/qps/rest/3.0/search/am/hostasset", f"{qualys_url}/qps/rest/2.0/search/am/hostasset",
json=payload, timeout=600) json=payload, timeout=600)
break break
except Exception: except Exception:
@ -913,7 +930,7 @@ def _fetch_asset_ids_by_tag(db, tag_name):
proxies = {"https": qualys_proxy, "http": qualys_proxy} if qualys_proxy else None proxies = {"https": qualys_proxy, "http": qualys_proxy} if qualys_proxy else None
try: try:
r = requests.post( r = requests.post(
f"{qualys_url}/qps/rest/3.0/search/am/hostasset", f"{qualys_url}/qps/rest/2.0/search/am/hostasset",
json={"ServiceRequest": { json={"ServiceRequest": {
"preferences": {"limitResults": 1000}, "preferences": {"limitResults": 1000},
"filters": {"Criteria": [ "filters": {"Criteria": [
@ -1156,7 +1173,7 @@ def fetch_all_qualys_assets(db, with_progress=False):
"filters": {"Criteria": criteria}}} "filters": {"Criteria": criteria}}}
try: try:
r = requests.post( r = requests.post(
f"{qualys_url}/qps/rest/3.0/search/am/hostasset", f"{qualys_url}/qps/rest/2.0/search/am/hostasset",
json=body, auth=(qualys_user, qualys_pass), json=body, auth=(qualys_user, qualys_pass),
verify=False, timeout=180, proxies=proxies, verify=False, timeout=180, proxies=proxies,
headers={"Content-Type": "application/json"} headers={"Content-Type": "application/json"}
@ -1264,7 +1281,7 @@ def delete_qualys_asset(db, asset_id):
proxies = {"https": qualys_proxy, "http": qualys_proxy} if qualys_proxy else None proxies = {"https": qualys_proxy, "http": qualys_proxy} if qualys_proxy else None
try: try:
r = requests.post( r = requests.post(
f"{qualys_url}/qps/rest/3.0/delete/am/hostasset/{int(asset_id)}", f"{qualys_url}/qps/rest/2.0/delete/am/hostasset/{int(asset_id)}",
auth=(qualys_user, qualys_pass), verify=False, timeout=60, proxies=proxies, auth=(qualys_user, qualys_pass), verify=False, timeout=60, proxies=proxies,
headers={"Content-Type": "application/json"} headers={"Content-Type": "application/json"}
) )

6
app/services/qualys_tags_service.py
View File

@ -70,7 +70,7 @@ def list_qualys_tags(db):
"""Liste tous les tags Qualys (nom, id, type DYN/STAT, ruleText).""" """Liste tous les tags Qualys (nom, id, type DYN/STAT, ruleText)."""
# search all tags # search all tags
payload = {"ServiceRequest": {"preferences": {"limitResults": 1000}}} payload = {"ServiceRequest": {"preferences": {"limitResults": 1000}}}
r = _qualys_post(db, "/qps/rest/3.0/search/am/tag", payload) r = _qualys_post(db, "/qps/rest/2.0/search/am/tag", payload)
if not r.get("ok"): if not r.get("ok"):
return {"ok": False, "msg": r.get("msg") or f"HTTP {r.get('status')}", "tags": []} return {"ok": False, "msg": r.get("msg") or f"HTTP {r.get('status')}", "tags": []}
tags = [] tags = []
@ -175,7 +175,7 @@ def create_static_tag(db, name, color="", description=""):
} }
} }
} }
r = _qualys_post(db, "/qps/rest/3.0/create/am/tag", payload) r = _qualys_post(db, "/qps/rest/2.0/create/am/tag", payload)
if r.get("ok"): if r.get("ok"):
return {"ok": True, "msg": f"Tag '{name}' cree"} return {"ok": True, "msg": f"Tag '{name}' cree"}
return {"ok": False, "msg": r.get("msg") or f"HTTP {r.get('status')}: {r.get('text', '')[:200]}"} return {"ok": False, "msg": r.get("msg") or f"HTTP {r.get('status')}: {r.get('text', '')[:200]}"}
@ -183,7 +183,7 @@ def create_static_tag(db, name, color="", description=""):
def delete_tag(db, tag_id): def delete_tag(db, tag_id):
"""Supprime un tag Qualys par id.""" """Supprime un tag Qualys par id."""
r = _qualys_post(db, f"/qps/rest/3.0/delete/am/tag/{tag_id}", {}) r = _qualys_post(db, f"/qps/rest/2.0/delete/am/tag/{tag_id}", {})
return {"ok": r.get("ok"), "msg": r.get("msg") or f"HTTP {r.get('status')}"} return {"ok": r.get("ok"), "msg": r.get("msg") or f"HTTP {r.get('status')}"}

2
app/templates/qualys_duplicates.html
View File

@ -45,7 +45,7 @@
<div class="card p-3"> <div class="card p-3">
<p class="text-xs text-gray-400 mb-2"> <p class="text-xs text-gray-400 mb-2">
Le plus recent (en haut de chaque groupe) est probablement l'asset actif. Les autres sont des zombies (anciennes installations, ré-IPs, doublons de scan). Le plus recent (en haut de chaque groupe) est probablement l'asset actif. Les autres sont des zombies (anciennes installations, ré-IPs, doublons de scan).
Bouton <strong>Supprimer</strong> = appel API Qualys <code>POST /qps/rest/3.0/delete/am/hostasset/{id}</code>. Bouton <strong>Supprimer</strong> = appel API Qualys <code>POST /qps/rest/2.0/delete/am/hostasset/{id}</code>.
{% if not can_delete %}<br><span class="text-yellow-400">Tu n'as pas la permission edit pour supprimer.</span>{% endif %} {% if not can_delete %}<br><span class="text-yellow-400">Tu n'as pas la permission edit pour supprimer.</span>{% endif %}
</p> </p>
<table class="w-full text-xs"> <table class="w-full text-xs">