new-site/scripts/workers/services/dot_esign.py

"""DOT/FMCSA e-signature helper.

Creates an `esign_records` row and emails the client a signing link, using the
local Postfix relay (localhost:25) like the other DOT status emails. This is the
DOT-side counterpart to telecom/esign_helper.py, but without the external
SMTP-auth dependency.

CRITICAL: portal links are signed with CUSTOMER_JWT_SECRET — the same secret the
API portal middleware (api/src/middleware/portalAuth.ts) verifies with. Signing
with any other secret yields an "Invalid portal link." error for the client.

Which DOT forms legally require the client's signature BEFORE we file:
  - mcs150 forms (MCS-150 biennial / new USDOT / reactivation) — perjury cert
  - operating authority (OP-1 / MC / emergency temporary authority) — applicant cert
  - carrier close-out (final MCS-150 "Out of Business") — perjury cert
  - entity dissolution (Articles of Dissolution) — no-liabilities attestation

UCR, drug-&-alcohol consortium enrollment, audit prep, and BOC-3 (signed by the
process agent, not the carrier) do NOT file a client-signed federal form, so they
skip signature capture entirely.
"""
from __future__ import annotations

import logging
import os

LOG = logging.getLogger("workers.services.dot_esign")

_MCS150_PERJURY = (
    "I certify under penalty of perjury that the information in this MCS-150 is "
    "true and correct to the best of my knowledge and belief, and that I am "
    "authorized to file it on behalf of the motor carrier. I understand that "
    "making a false statement is punishable under 18 U.S.C. § 1001."
)
_OP1_CERT = (
    "I certify that I am authorized to apply for operating authority on behalf of "
    "the applicant, that the information provided is true and correct, and that the "
    "applicant will comply with all applicable FMCSA safety and insurance "
    "requirements. I understand that a false statement is punishable under "
    "18 U.S.C. § 1001."
)
_CLOSEOUT_CERT = (
    "I authorize Performance West Inc. to file a final MCS-150 marking this carrier "
    "OUT OF BUSINESS and to deactivate the USDOT number and operating authority "
    "listed above. I certify under penalty of perjury that the information is true "
    "and correct and that I am authorized to wind down this motor carrier."
)
_DISSOLUTION_ATTEST = (
    "I attest that, to the best of my knowledge, this entity has NO outstanding "
    "lawsuits, liens, judgments, or unsatisfied creditor claims. I authorize "
    "Performance West Inc. to file Articles of Dissolution and final returns on the "
    "entity's behalf, and I certify that I am authorized to dissolve this entity. I "
    "understand that dissolving an entity with unresolved liabilities may expose its "
    "members or officers to personal liability."
)

# slug -> signing config. A slug absent from this map requires NO signature.
DOT_SIGNING: dict[str, dict] = {
    "mcs150-update": {
        "document_type": "mcs150",
        "document_title": "MCS-150 Biennial Update — Certification Under Penalty of Perjury",
        "perjury_text": _MCS150_PERJURY,
    },
    "dot-registration": {
        "document_type": "mcs150",
        "document_title": "New USDOT Registration (MCS-150) — Certification Under Penalty of Perjury",
        "perjury_text": _MCS150_PERJURY,
    },
    "usdot-reactivation": {
        "document_type": "mcs150",
        "document_title": "USDOT Reactivation (MCS-150) — Certification Under Penalty of Perjury",
        "perjury_text": _MCS150_PERJURY,
    },
    "dot-full-compliance": {
        "document_type": "mcs150",
        "document_title": "DOT Full Compliance (MCS-150) — Certification Under Penalty of Perjury",
        "perjury_text": _MCS150_PERJURY,
    },
    "mc-authority": {
        "document_type": "operating-authority",
        "document_title": "Operating Authority Application (OP-1) — Applicant Certification",
        "perjury_text": _OP1_CERT,
    },
    "emergency-temporary-authority": {
        "document_type": "operating-authority",
        "document_title": "Emergency Temporary Authority Request — Applicant Certification",
        "perjury_text": _OP1_CERT,
    },
    "carrier-closeout": {
        "document_type": "carrier-closeout",
        "document_title": "Carrier Close-Out Authorization & Final MCS-150 (Out of Business)",
        "perjury_text": _CLOSEOUT_CERT,
    },
    "entity-dissolution": {
        "document_type": "entity-dissolution",
        "document_title": "Entity Dissolution Authorization & No-Liabilities Attestation",
        "perjury_text": _DISSOLUTION_ATTEST,
    },
}


def requires_signature(slug: str) -> bool:
    """True if the DOT service files a form that needs the client's signature."""
    return slug in DOT_SIGNING


# AcroForm signature field name -> logical anchor field, per document_type.
# The stamper grows the box upward so the signature rests on the official rule.
DOC_SIG_FIELDS: dict[str, dict[str, str]] = {
    "mcs150": {"certifySignature": "signer"},
    "carrier-closeout": {"certifySignature": "signer"},
}


def _extract_acroform_anchors(document_minio_key: str, document_type: str) -> list | None:
    """Download the unsigned form from MinIO and return signature anchors.

    Returns a list of anchor dicts (placed on the official signature line) for
    document types backed by a real AcroForm PDF, or None if not applicable / on
    any failure (the signing flow still works without anchors, falling back to a
    typed-name overlay only if anchors are present).
    """
    field_map = DOC_SIG_FIELDS.get(document_type)
    if not field_map or not document_minio_key:
        return None
    try:
        import tempfile

        from .signature_stamper import anchors_from_acroform

        try:
            from scripts.document_gen.minio_client import MinioStorage
        except ImportError:
            from document_gen.minio_client import MinioStorage  # type: ignore

        storage = MinioStorage()
        with tempfile.NamedTemporaryFile(suffix=".pdf", delete=True) as tf:
            storage.download(document_minio_key, tf.name)
            tf.seek(0)
            pdf_bytes = open(tf.name, "rb").read()
        anchors = anchors_from_acroform(pdf_bytes, field_map)
        return anchors or None
    except Exception as exc:
        LOG.warning("Could not extract signature anchors from %s: %s", document_minio_key, exc)
        return None


def request_dot_esign(
    order_number: str,
    slug: str,
    entity_name: str,
    customer_email: str,
    dot_number: str = "",
    document_minio_key: str = "",
    extra_metadata: dict | None = None,
    expires_days: int = 7,
) -> int | None:
    """Create a pending esign record for a DOT form and email the signing link.

    Idempotent per (order_number, document_type): re-running will not duplicate a
    pending/signed record. Returns the esign_records.id, or None on failure.
    """
    cfg = DOT_SIGNING.get(slug)
    if not cfg:
        return None  # this service does not require a signature
    if not customer_email:
        LOG.warning("[%s] No customer email — cannot request signature", order_number)
        return None

    import json

    document_type = cfg["document_type"]
    document_title = cfg["document_title"]
    perjury_text = cfg["perjury_text"]

    metadata = {
        "dot_number": dot_number,
        "service_slug": slug,
        "perjury_text": perjury_text,
    }
    if extra_metadata:
        metadata.update(extra_metadata)

    # Extract the exact signature-line coordinates from the unsigned form so the
    # client's signature can be stamped right on the rule after they sign.
    anchors = _extract_acroform_anchors(document_minio_key, document_type)

    # 1. Upsert the pending record
    esign_id = None
    try:
        import psycopg2

        conn = psycopg2.connect(os.environ.get("DATABASE_URL", ""))
        with conn.cursor() as cur:
            cur.execute(
                """
                INSERT INTO esign_records (
                    order_number, document_type, document_title, entity_name,
                    document_minio_key, document_metadata, signature_anchors,
                    requires_perjury, status, expires_at
                ) VALUES (%s, %s, %s, %s, %s, %s, %s, TRUE, 'pending',
                          NOW() + (%s || ' days')::interval)
                ON CONFLICT (order_number, document_type)
                    WHERE status IN ('pending', 'signed')
                DO UPDATE SET
                    document_title = EXCLUDED.document_title,
                    entity_name = EXCLUDED.entity_name,
                    document_minio_key = EXCLUDED.document_minio_key,
                    document_metadata = EXCLUDED.document_metadata,
                    signature_anchors = EXCLUDED.signature_anchors,
                    expires_at = EXCLUDED.expires_at,
                    updated_at = NOW()
                RETURNING id
                """,
                (
                    order_number, document_type, document_title, entity_name,
                    document_minio_key, json.dumps(metadata),
                    json.dumps(anchors) if anchors else None, str(expires_days),
                ),
            )
            row = cur.fetchone()
            esign_id = row[0] if row else None
        conn.commit()
        conn.close()
    except Exception as exc:
        LOG.error("[%s] Failed to create esign record (%s): %s", order_number, document_type, exc)
        return None

    # 2. Email the signing link (signed with CUSTOMER_JWT_SECRET to match the portal)
    try:
        _send_signing_email(
            order_number=order_number,
            document_type=document_type,
            document_title=document_title,
            entity_name=entity_name,
            dot_number=dot_number,
            customer_email=customer_email,
            expires_days=expires_days,
        )
        LOG.info("[%s] Signing link sent to %s (%s)", order_number, customer_email, document_type)
    except Exception as exc:
        LOG.warning("[%s] Could not send signing email (record exists): %s", order_number, exc)

    return esign_id


def _send_signing_email(
    order_number: str,
    document_type: str,
    document_title: str,
    entity_name: str,
    dot_number: str,
    customer_email: str,
    expires_days: int,
) -> None:
    import smtplib
    from email.mime.text import MIMEText

    try:
        import jwt as pyjwt
    except ImportError:  # pragma: no cover
        import PyJWT as pyjwt  # type: ignore

    secret = os.environ.get("CUSTOMER_JWT_SECRET", "changeme_long_random_string")
    domain = os.environ.get("DOMAIN", "performancewest.net")

    from datetime import datetime, timedelta, timezone

    token = pyjwt.encode(
        {
            "order_id": order_number,
            "order_type": document_type,
            "email": customer_email,
            "exp": datetime.now(timezone.utc) + timedelta(days=expires_days),
        },
        secret,
        algorithm="HS256",
    )
    sign_url = f"https://{domain}/portal/esign?token={token}"

    dot_line = f" (DOT# {dot_number})" if dot_number else ""
    body = (
        f"Hi,\n\n"
        f"Your {document_title} for {entity_name}{dot_line} has been prepared and "
        f"is ready for your signature.\n\n"
        f"Federal law requires your certification before we can submit this filing. "
        f"Please review and sign here:\n{sign_url}\n\n"
        f"This link expires in {expires_days} days.\n\n"
        f"Once you sign, we file with the appropriate agency and send you "
        f"confirmation.\n\n"
        f"Order: {order_number}\n"
        f"Questions? Call (888) 411-0383.\n\n"
        f"Performance West Inc.\n"
    )

    msg = MIMEText(body)
    msg["Subject"] = f"Action Required: Sign Your {document_title} — {entity_name}"
    msg["From"] = "noreply@performancewest.net"
    msg["To"] = customer_email

    with smtplib.SMTP("localhost", 25, timeout=30) as s:
        s.sendmail(msg["From"], [customer_email], msg.as_string())