99 lines
4.3 KiB
Python
99 lines
4.3 KiB
Python
import math
|
|
|
|
def rule_missing_status_after_boot(frames, ctx):
|
|
if len(frames) < 3: return None
|
|
has_boot = any(f['action'] == "BootNotification" for f in frames)
|
|
has_status = any(f['action'] == "StatusNotification" for f in frames)
|
|
if has_boot and not has_status:
|
|
return {
|
|
"id": "NO_STATUS_AFTER_BOOT", "severity": "warning", "confidence": "medium",
|
|
"title": "Séquence de démarrage non conforme",
|
|
"explication": "Aucun StatusNotification n'a été reçu après le BootNotification.",
|
|
"action": "Envoyez un TriggerMessage(StatusNotification)."
|
|
}
|
|
return None
|
|
|
|
def rule_bad_websocket_ping(frames, ctx):
|
|
for f in reversed(frames):
|
|
if f['action'] == "GetConfiguration" and f['direction'] == "OUT":
|
|
config = f['payload'].get('configurationKey', [])
|
|
for item in config:
|
|
if item.get('key') == 'WebSocketPingInterval' and item.get('value') == '0':
|
|
return {
|
|
"id": "BAD_PING_INTERVAL", "severity": "critical", "confidence": "high",
|
|
"title": "WebSocketPingInterval désactivé (0)",
|
|
"explication": "La borne n'envoie pas de ping. Derrière une box FAI (NAT), la connexion sera coupée.",
|
|
"action": "Corrigez cette valeur à 60 depuis l'onglet Audit Config."
|
|
}
|
|
return None
|
|
|
|
def check_electrical(borne_config_dict, installer_config):
|
|
findings = []
|
|
max_c_str = borne_config_dict.get("MaxCurrent", {}).get("value")
|
|
disj_str = str(installer_config.get("disj_calibre", "0"))
|
|
p_decl_str = str(installer_config.get("borne_puissance", "0"))
|
|
raccordement = installer_config.get("raccordement", "")
|
|
|
|
if not (max_c_str and disj_str and p_decl_str): return findings
|
|
|
|
try:
|
|
max_c = int(max_c_str)
|
|
disj = int(disj_str)
|
|
p_decl = int(p_decl_str)
|
|
|
|
# a) Sécurité disjoncteur
|
|
if disj < max_c:
|
|
findings.append({
|
|
"id": "ELEC_UNDER_PROTECTED", "severity": "critical", "confidence": "high",
|
|
"title": "Protection sous-dimensionnée",
|
|
"explication": f"Disjoncteur {disj}A inférieur au courant max borne {max_c}A.",
|
|
"action": "Risque de déclenchement en charge. Vérifier le calibre et la section de câble."
|
|
})
|
|
|
|
# b) Puissance réelle vs déclarée
|
|
if "mono" in raccordement:
|
|
p_reelle = 230 * max_c
|
|
else:
|
|
p_reelle = 400 * max_c * math.sqrt(3)
|
|
|
|
p_reelle_kw = round(p_reelle / 1000, 1)
|
|
|
|
if p_reelle_kw < p_decl * 0.85:
|
|
findings.append({
|
|
"id": "ELEC_POWER_BRIDLED", "severity": "critical", "confidence": "high",
|
|
"title": "Borne bridée logiquement",
|
|
"explication": f"Borne bridée à {p_reelle_kw} kW ({max_c}A x {400 if 'tri' in raccordement else 230}V) alors que {p_decl} kW sont déclarés.",
|
|
"action": "Le firmware limite le courant. Vérifier le paramétrage usine ou une limitation logicielle."
|
|
})
|
|
|
|
# c) 22kW sur Mono
|
|
if p_decl > 7.4 and "mono" in raccordement:
|
|
findings.append({
|
|
"id": "ELEC_MONO_OVERPOWER", "severity": "critical", "confidence": "high",
|
|
"title": "Raccordement incohérent",
|
|
"explication": f"Puissance {p_decl}kW déclarée sur du monophasé.",
|
|
"action": "Au-delà de 7.4kW, le triphasé est obligatoire."
|
|
})
|
|
except ValueError:
|
|
pass
|
|
return findings
|
|
|
|
def analyze_frames(frames, cp_data, installer_config=None):
|
|
findings = []
|
|
|
|
# Rècles comportementales
|
|
rules = [rule_missing_status_after_boot, rule_bad_websocket_ping]
|
|
for rule in rules:
|
|
res = rule(frames, cp_data)
|
|
if res: findings.append(res)
|
|
|
|
# Règles électriques (nécessitent la config installateur)
|
|
if installer_config:
|
|
borne_config_dict = {}
|
|
for f in reversed(frames):
|
|
if f['action'] == "GetConfiguration" and f['direction'] == "OUT":
|
|
borne_config_dict = {k["key"]: k for k in f['payload'].get('configurationKey', [])}
|
|
break
|
|
findings.extend(check_electrical(borne_config_dict, installer_config))
|
|
|
|
return findings
|