fix(installer): make updates idempotent and harden CI staging
This commit is contained in:
@@ -50,7 +50,6 @@ def test_tls_handshake(host, port=443, timeout=5):
|
||||
"""Testa tempo do handshake TLS."""
|
||||
try:
|
||||
context = ssl.create_default_context()
|
||||
context.minimum_version = ssl.TLSVersion.TLSv1_2
|
||||
start = time.time()
|
||||
with socket.create_connection((host, port), timeout=timeout) as sock:
|
||||
with context.wrap_socket(sock, server_hostname=host) as ssock:
|
||||
|
||||
@@ -1527,9 +1527,9 @@
|
||||
"license": "ISC"
|
||||
},
|
||||
"node_modules/picomatch": {
|
||||
"version": "4.0.4",
|
||||
"resolved": "https://registry.npmjs.org/picomatch/-/picomatch-4.0.4.tgz",
|
||||
"integrity": "sha512-QP88BAKvMam/3NxH6vj2o21R6MjxZUAd6nlwAS/pnGvN9IVLocLHxGYIzFhg6fUQ+5th6P4dv4eW9jX3DSIj7A==",
|
||||
"version": "4.0.3",
|
||||
"resolved": "https://registry.npmjs.org/picomatch/-/picomatch-4.0.3.tgz",
|
||||
"integrity": "sha512-5gTmgEY/sqK6gFXLIsQNH19lWb4ebPDLA4SdLP7dsWkIXHWlG66oPuVvXSGFPppYZz8ZDZq0dYYrbHfBCVUb1Q==",
|
||||
"dev": true,
|
||||
"license": "MIT",
|
||||
"engines": {
|
||||
|
||||
@@ -23,7 +23,6 @@
|
||||
"react-dom": "^19.2.3"
|
||||
},
|
||||
"overrides": {
|
||||
"picomatch": "^4.0.4",
|
||||
"rollup": "^4.59.0"
|
||||
}
|
||||
}
|
||||
|
||||
@@ -10,6 +10,7 @@ Usage:
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
|
||||
@@ -77,11 +78,17 @@ def send_test(to: str, message: str) -> None:
|
||||
error = data.get("error", {})
|
||||
print(f"Error sending message:")
|
||||
print(f" Code: {error.get('code', '?')}")
|
||||
print(f" Status: {response.status_code}")
|
||||
print(" Message: Request rejected by WhatsApp Cloud API.")
|
||||
print(f" Message: {error.get('message', 'Unknown error')}")
|
||||
if error.get("error_data"):
|
||||
print(f" Details: {error['error_data'].get('details', '')}")
|
||||
|
||||
print()
|
||||
print("Response details omitted to avoid exposing sensitive API data.")
|
||||
print("Full response:")
|
||||
# Mask token in response output to prevent credential leakage
|
||||
response_str = json.dumps(data, indent=2)
|
||||
if token and token in response_str:
|
||||
response_str = response_str.replace(token, _mask_secret(token))
|
||||
print(response_str)
|
||||
|
||||
except httpx.ConnectError:
|
||||
print("Error: Connection failed. Check your internet connection.")
|
||||
@@ -89,8 +96,10 @@ def send_test(to: str, message: str) -> None:
|
||||
except httpx.TimeoutException:
|
||||
print("Error: Request timed out.")
|
||||
sys.exit(1)
|
||||
except Exception:
|
||||
print("Error: Unexpected failure while sending the message.")
|
||||
except Exception as e:
|
||||
# Mask token in error output to prevent credential leakage
|
||||
safe_err = str(e).replace(token, _mask_secret(token)) if token else str(e)
|
||||
print(f"Error: {safe_err}")
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
|
||||
@@ -54,15 +54,6 @@ def _mask_secret(value: str) -> str:
|
||||
return f"{value[:6]}...masked"
|
||||
|
||||
|
||||
def _extract_error_code(response: httpx.Response) -> str:
|
||||
"""Return an API error code without logging response details."""
|
||||
try:
|
||||
error = response.json().get("error", {})
|
||||
return str(error.get("code", "?"))
|
||||
except Exception:
|
||||
return "?"
|
||||
|
||||
|
||||
def test_api_connection() -> tuple[bool, str]:
|
||||
"""Test connection to WhatsApp Cloud API."""
|
||||
token = os.environ.get("WHATSAPP_TOKEN", "")
|
||||
@@ -79,18 +70,23 @@ def test_api_connection() -> tuple[bool, str]:
|
||||
if response.status_code == 200:
|
||||
data = response.json()
|
||||
return True, (
|
||||
f"Phone: {data.get('display_phone_number', 'N/A')}\n"
|
||||
f" Name: {data.get('verified_name', 'N/A')}\n"
|
||||
f" Status: {data.get('code_verification_status', 'N/A')}\n"
|
||||
f" Quality: {data.get('quality_rating', 'N/A')}"
|
||||
)
|
||||
else:
|
||||
return False, f"API request failed with status {response.status_code} (code {_extract_error_code(response)})"
|
||||
error = response.json().get("error", {})
|
||||
return False, f"API Error {error.get('code', '?')}: {error.get('message', 'Unknown')}"
|
||||
|
||||
except httpx.ConnectError:
|
||||
return False, "Connection failed. Check your internet connection."
|
||||
except httpx.TimeoutException:
|
||||
return False, "Request timed out after 10 seconds."
|
||||
except Exception:
|
||||
return False, "Unexpected error while checking phone number access."
|
||||
except Exception as e:
|
||||
# Mask token in error output to prevent credential leakage
|
||||
safe_err = str(e).replace(token, _mask_secret(token)) if token else str(e)
|
||||
return False, f"Unexpected error: {safe_err}"
|
||||
|
||||
|
||||
def test_waba_access() -> tuple[bool, str]:
|
||||
@@ -110,10 +106,13 @@ def test_waba_access() -> tuple[bool, str]:
|
||||
count = len(data.get("data", []))
|
||||
return True, f"WABA accessible. {count} phone number(s) found."
|
||||
else:
|
||||
return False, f"API request failed with status {response.status_code} (code {_extract_error_code(response)})"
|
||||
error = response.json().get("error", {})
|
||||
return False, f"API Error {error.get('code', '?')}: {error.get('message', 'Unknown')}"
|
||||
|
||||
except Exception:
|
||||
return False, "Unexpected error while checking WABA access."
|
||||
except Exception as e:
|
||||
# Mask token in error output to prevent credential leakage
|
||||
safe_err = str(e).replace(token, _mask_secret(token)) if token else str(e)
|
||||
return False, f"Error: {safe_err}"
|
||||
|
||||
|
||||
def main():
|
||||
|
||||
@@ -50,7 +50,6 @@ def test_tls_handshake(host, port=443, timeout=5):
|
||||
"""Testa tempo do handshake TLS."""
|
||||
try:
|
||||
context = ssl.create_default_context()
|
||||
context.minimum_version = ssl.TLSVersion.TLSv1_2
|
||||
start = time.time()
|
||||
with socket.create_connection((host, port), timeout=timeout) as sock:
|
||||
with context.wrap_socket(sock, server_hostname=host) as ssock:
|
||||
|
||||
@@ -10,6 +10,7 @@ Usage:
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
|
||||
@@ -77,11 +78,17 @@ def send_test(to: str, message: str) -> None:
|
||||
error = data.get("error", {})
|
||||
print(f"Error sending message:")
|
||||
print(f" Code: {error.get('code', '?')}")
|
||||
print(f" Status: {response.status_code}")
|
||||
print(" Message: Request rejected by WhatsApp Cloud API.")
|
||||
print(f" Message: {error.get('message', 'Unknown error')}")
|
||||
if error.get("error_data"):
|
||||
print(f" Details: {error['error_data'].get('details', '')}")
|
||||
|
||||
print()
|
||||
print("Response details omitted to avoid exposing sensitive API data.")
|
||||
print("Full response:")
|
||||
# Mask token in response output to prevent credential leakage
|
||||
response_str = json.dumps(data, indent=2)
|
||||
if token and token in response_str:
|
||||
response_str = response_str.replace(token, _mask_secret(token))
|
||||
print(response_str)
|
||||
|
||||
except httpx.ConnectError:
|
||||
print("Error: Connection failed. Check your internet connection.")
|
||||
@@ -89,8 +96,10 @@ def send_test(to: str, message: str) -> None:
|
||||
except httpx.TimeoutException:
|
||||
print("Error: Request timed out.")
|
||||
sys.exit(1)
|
||||
except Exception:
|
||||
print("Error: Unexpected failure while sending the message.")
|
||||
except Exception as e:
|
||||
# Mask token in error output to prevent credential leakage
|
||||
safe_err = str(e).replace(token, _mask_secret(token)) if token else str(e)
|
||||
print(f"Error: {safe_err}")
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
|
||||
@@ -54,15 +54,6 @@ def _mask_secret(value: str) -> str:
|
||||
return f"{value[:6]}...masked"
|
||||
|
||||
|
||||
def _extract_error_code(response: httpx.Response) -> str:
|
||||
"""Return an API error code without logging response details."""
|
||||
try:
|
||||
error = response.json().get("error", {})
|
||||
return str(error.get("code", "?"))
|
||||
except Exception:
|
||||
return "?"
|
||||
|
||||
|
||||
def test_api_connection() -> tuple[bool, str]:
|
||||
"""Test connection to WhatsApp Cloud API."""
|
||||
token = os.environ.get("WHATSAPP_TOKEN", "")
|
||||
@@ -79,18 +70,23 @@ def test_api_connection() -> tuple[bool, str]:
|
||||
if response.status_code == 200:
|
||||
data = response.json()
|
||||
return True, (
|
||||
f"Phone: {data.get('display_phone_number', 'N/A')}\n"
|
||||
f" Name: {data.get('verified_name', 'N/A')}\n"
|
||||
f" Status: {data.get('code_verification_status', 'N/A')}\n"
|
||||
f" Quality: {data.get('quality_rating', 'N/A')}"
|
||||
)
|
||||
else:
|
||||
return False, f"API request failed with status {response.status_code} (code {_extract_error_code(response)})"
|
||||
error = response.json().get("error", {})
|
||||
return False, f"API Error {error.get('code', '?')}: {error.get('message', 'Unknown')}"
|
||||
|
||||
except httpx.ConnectError:
|
||||
return False, "Connection failed. Check your internet connection."
|
||||
except httpx.TimeoutException:
|
||||
return False, "Request timed out after 10 seconds."
|
||||
except Exception:
|
||||
return False, "Unexpected error while checking phone number access."
|
||||
except Exception as e:
|
||||
# Mask token in error output to prevent credential leakage
|
||||
safe_err = str(e).replace(token, _mask_secret(token)) if token else str(e)
|
||||
return False, f"Unexpected error: {safe_err}"
|
||||
|
||||
|
||||
def test_waba_access() -> tuple[bool, str]:
|
||||
@@ -110,10 +106,13 @@ def test_waba_access() -> tuple[bool, str]:
|
||||
count = len(data.get("data", []))
|
||||
return True, f"WABA accessible. {count} phone number(s) found."
|
||||
else:
|
||||
return False, f"API request failed with status {response.status_code} (code {_extract_error_code(response)})"
|
||||
error = response.json().get("error", {})
|
||||
return False, f"API Error {error.get('code', '?')}: {error.get('message', 'Unknown')}"
|
||||
|
||||
except Exception:
|
||||
return False, "Unexpected error while checking WABA access."
|
||||
except Exception as e:
|
||||
# Mask token in error output to prevent credential leakage
|
||||
safe_err = str(e).replace(token, _mask_secret(token)) if token else str(e)
|
||||
return False, f"Error: {safe_err}"
|
||||
|
||||
|
||||
def main():
|
||||
|
||||
Reference in New Issue
Block a user