135 lines
5.6 KiB
Python
135 lines
5.6 KiB
Python
import paramiko
|
|
import os
|
|
import urllib.request
|
|
import json
|
|
import ssl
|
|
import sys
|
|
import io
|
|
import time
|
|
import tarfile
|
|
|
|
if sys.platform == 'win32':
|
|
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8', errors='replace')
|
|
sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding='utf-8', errors='replace')
|
|
|
|
HOST = "161.97.93.156"
|
|
USER = "carlos"
|
|
PASSWORD = "Innova1978+*?"
|
|
REMOTE_TEMP_TAR = "/home/carlos/pacifictrans_otp_update.tar.gz"
|
|
|
|
LOCAL_BACKEND = r"c:\Users\Carlos\dev\almeira-web\transporte-cali-buenaventura\backend"
|
|
LOCAL_APK = r"c:\Users\Carlos\dev\almeira-web\pacifictrans-app\public\PacifictransGo.apk"
|
|
LOCAL_TAR = r"c:\Users\Carlos\dev\almeira-web\pacifictrans_otp_update.tar.gz"
|
|
|
|
API_URL = "https://pacifictrans.almeira.com.co"
|
|
|
|
ctx = ssl.create_default_context()
|
|
ctx.check_hostname = False
|
|
ctx.verify_mode = ssl.CERT_NONE
|
|
|
|
def create_bundle_tar():
|
|
print("[1/4] Creando empaquetado de actualización de seguridad backend & frontend...", flush=True)
|
|
if os.path.exists(LOCAL_TAR):
|
|
os.remove(LOCAL_TAR)
|
|
|
|
with tarfile.open(LOCAL_TAR, "w:gz") as tar:
|
|
dist_path = os.path.join(LOCAL_BACKEND, "dist")
|
|
if os.path.exists(dist_path):
|
|
tar.add(dist_path, arcname="dist")
|
|
public_path = os.path.join(LOCAL_BACKEND, "public")
|
|
if os.path.exists(public_path):
|
|
tar.add(public_path, arcname="public")
|
|
data_path = os.path.join(LOCAL_BACKEND, "data")
|
|
if os.path.exists(data_path):
|
|
tar.add(data_path, arcname="data")
|
|
mig_path = os.path.join(LOCAL_BACKEND, "migrations")
|
|
if os.path.exists(mig_path):
|
|
tar.add(mig_path, arcname="migrations")
|
|
|
|
size_mb = os.path.getsize(LOCAL_TAR) / (1024 * 1024)
|
|
print(f" -> Archivo empaquetado creado: {LOCAL_TAR} ({size_mb:.2f} MB)", flush=True)
|
|
|
|
def deploy_to_vps():
|
|
print("[2/4] Conectando por SSH al VPS y desplegando actualización en contenedor Coolify...", flush=True)
|
|
ssh = paramiko.SSHClient()
|
|
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
|
|
ssh.connect(hostname=HOST, username=USER, password=PASSWORD, timeout=30)
|
|
|
|
sftp = ssh.open_sftp()
|
|
print(f" -> Subiendo {REMOTE_TEMP_TAR} al VPS...", flush=True)
|
|
sftp.put(LOCAL_TAR, REMOTE_TEMP_TAR)
|
|
|
|
if os.path.exists(LOCAL_APK):
|
|
print(" -> Subiendo APK actualizado al VPS...", flush=True)
|
|
sftp.put(LOCAL_APK, "/home/carlos/apps/dripzone-app/public/PacifictransGo.apk")
|
|
sftp.close()
|
|
|
|
stdin, stdout, stderr = ssh.exec_command("docker ps --format '{{.Names}}'")
|
|
containers = [c.strip() for c in stdout.read().decode('utf-8').splitlines() if c.strip()]
|
|
pacific_containers = [c for c in containers if any(k in c.lower() for k in ['tayk', 'pacific'])]
|
|
|
|
if not pacific_containers:
|
|
pacific_containers = [c for c in containers if 'tayk' in c.lower()]
|
|
|
|
print(f" -> Contenedores destino en Coolify: {pacific_containers}", flush=True)
|
|
|
|
for c in pacific_containers:
|
|
print(f" -> Actualizando contenedor: {c}...", flush=True)
|
|
ssh.exec_command(f"docker cp {REMOTE_TEMP_TAR} {c}:/app/pacifictrans_otp_update.tar.gz")
|
|
stdin, stdout, stderr = ssh.exec_command(
|
|
f"docker exec {c} sh -c 'cd /app && tar -xzf pacifictrans_otp_update.tar.gz && rm -f pacifictrans_otp_update.tar.gz'"
|
|
)
|
|
stdout.channel.recv_exit_status()
|
|
|
|
if os.path.exists(LOCAL_APK):
|
|
ssh.exec_command(f"docker cp /home/carlos/apps/dripzone-app/public/PacifictransGo.apk {c}:/app/public/PacifictransGo.apk")
|
|
|
|
print(f" -> Reiniciando contenedor {c}...", flush=True)
|
|
stdin, stdout, stderr = ssh.exec_command(f"docker restart {c}")
|
|
stdout.channel.recv_exit_status()
|
|
print(f" [OK] Contenedor {c} reiniciado con el parche de seguridad y WhatsApp activo.", flush=True)
|
|
|
|
ssh.exec_command(f"rm -f {REMOTE_TEMP_TAR}")
|
|
ssh.close()
|
|
|
|
def verify_live():
|
|
print("[3/4] Verificando endpoints y envío OTP seguro en producción...", flush=True)
|
|
time.sleep(6)
|
|
|
|
try:
|
|
req = urllib.request.Request(f"{API_URL}/api/health")
|
|
with urllib.request.urlopen(req, context=ctx, timeout=15) as res:
|
|
data = json.loads(res.read().decode('utf-8'))
|
|
print(f" [OK] Healthcheck producción: {data.get('status')} (HTTP {res.status})", flush=True)
|
|
except Exception as e:
|
|
print(f" [Aviso Healthcheck]: {e}", flush=True)
|
|
|
|
# Prueba de envío OTP seguro (no debe retornar codePreview)
|
|
print(" -> Verificando que /api/auth/send-otp NO exponga el código OTP en la respuesta...", flush=True)
|
|
try:
|
|
req = urllib.request.Request(
|
|
f"{API_URL}/api/auth/send-otp",
|
|
data=json.dumps({"identifier": "3167455877"}).encode("utf-8"),
|
|
headers={"Content-Type": "application/json"}
|
|
)
|
|
with urllib.request.urlopen(req, context=ctx, timeout=20) as res:
|
|
data = json.loads(res.read().decode("utf-8"))
|
|
has_leak = "codePreview" in data or "code" in data
|
|
print(f" [Respuesta Producción]: HTTP {res.status}, success: {data.get('success')}, maskedPhone: {data.get('maskedPhone')}")
|
|
if has_leak:
|
|
print(f" [ALERTA DE SEGURIDAD]: La respuesta aún contiene el código: {data}", flush=True)
|
|
else:
|
|
print(" [SEGURIDAD CONFIRMADA]: La respuesta NO contiene el código OTP. Entrega 100% por WhatsApp.", flush=True)
|
|
except Exception as e:
|
|
print(f" [Error en prueba de OTP]: {e}", flush=True)
|
|
|
|
print("\n[4/4] Proceso finalizado.")
|
|
|
|
def main():
|
|
create_bundle_tar()
|
|
deploy_to_vps()
|
|
verify_live()
|
|
|
|
if __name__ == '__main__':
|
|
main()
|