add fax filing pipeline: VitalPBX sender, attestation cover page with digital signature, compliance checker pending filing override
- filing_attestation.py: generates cover page attesting PW submitted document to recipient with date/time stamp, contact info, and digital signature - fax_sender.py: sends PDFs via VitalPBX API, polls for delivery, generates attested copy for customer records - dot-lookup.ts: if DOT has pending MCS-150 order, show green 'UPDATE SUBMITTED' instead of red 'OVERDUE' in compliance checker - requirements.txt: add pyhanko + cryptography for PDF digital signatures
This commit is contained in:
justin
parent
e2c7cc582b
commit
1f1113d63c
4 changed files with 645 additions and 1 deletions
355
scripts/document_gen/templates/filing_attestation.py
Normal file
355
scripts/document_gen/templates/filing_attestation.py
Normal file
|
|
@ -0,0 +1,355 @@
|
|||
"""Filing Attestation Cover Page Generator.
|
||||
|
||||
After a fax is confirmed delivered, this generates a cover page attesting
|
||||
that Performance West submitted the document to the recipient, with a
|
||||
date/time stamp and contact information for regulatory verification.
|
||||
|
||||
The cover page is prepended to the original filed PDF and stored as the
|
||||
customer's proof of filing.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import os
|
||||
import tempfile
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
|
||||
LOG = logging.getLogger("document_gen.filing_attestation")
|
||||
|
||||
try:
|
||||
from pypdf import PdfReader, PdfWriter
|
||||
from pypdf.generic import NameObject
|
||||
except ImportError:
|
||||
PdfReader = None
|
||||
|
||||
|
||||
def generate_attestation_page(
|
||||
order_number: str,
|
||||
dot_number: str,
|
||||
entity_name: str,
|
||||
document_type: str,
|
||||
submitted_at: datetime,
|
||||
recipient_name: str = "Federal Motor Carrier Safety Administration (FMCSA)",
|
||||
) -> str:
|
||||
"""Generate a one-page attestation PDF.
|
||||
|
||||
Args:
|
||||
order_number: Our internal order number.
|
||||
dot_number: USDOT number on the filing.
|
||||
entity_name: Legal entity name.
|
||||
document_type: e.g. "MCS-150 Biennial Update", "MCS-150B", etc.
|
||||
submitted_at: UTC datetime when the filing was transmitted.
|
||||
recipient_name: Who it was sent to (don't include fax/phone numbers).
|
||||
|
||||
Returns:
|
||||
Path to the attestation PDF.
|
||||
"""
|
||||
from reportlab.lib.pagesizes import letter
|
||||
from reportlab.lib.units import inch
|
||||
from reportlab.pdfgen import canvas
|
||||
from reportlab.lib.colors import HexColor
|
||||
|
||||
work_dir = tempfile.mkdtemp(prefix="pw_attest_")
|
||||
filepath = os.path.join(work_dir, f"attestation_{order_number}.pdf")
|
||||
|
||||
c = canvas.Canvas(filepath, pagesize=letter)
|
||||
w, h = letter
|
||||
|
||||
# Header bar
|
||||
c.setFillColor(HexColor("#1a2744"))
|
||||
c.rect(0, h - 1.2 * inch, w, 1.2 * inch, fill=True, stroke=False)
|
||||
|
||||
c.setFillColor(HexColor("#ffffff"))
|
||||
c.setFont("Helvetica-Bold", 20)
|
||||
c.drawCentredString(w / 2, h - 0.6 * inch, "CERTIFICATE OF FILING")
|
||||
c.setFont("Helvetica", 11)
|
||||
c.drawCentredString(w / 2, h - 0.9 * inch, "Performance West Inc. — DOT Compliance Services")
|
||||
|
||||
# Body
|
||||
y = h - 1.8 * inch
|
||||
c.setFillColor(HexColor("#1a2744"))
|
||||
|
||||
def line(text, font="Helvetica", size=12, spacing=22):
|
||||
nonlocal y
|
||||
c.setFont(font, size)
|
||||
c.drawString(1 * inch, y, text)
|
||||
y -= spacing
|
||||
|
||||
def label_value(label, value, spacing=22):
|
||||
nonlocal y
|
||||
c.setFont("Helvetica-Bold", 11)
|
||||
c.drawString(1 * inch, y, label)
|
||||
c.setFont("Helvetica", 11)
|
||||
c.drawString(3.2 * inch, y, str(value))
|
||||
y -= spacing
|
||||
|
||||
# Attestation text
|
||||
c.setFont("Helvetica", 12)
|
||||
text_block = (
|
||||
f"This is to certify that Performance West Inc., acting as the authorized "
|
||||
f"compliance representative for {entity_name}, submitted the following "
|
||||
f"document to {recipient_name} on the date and time indicated below."
|
||||
)
|
||||
|
||||
# Word wrap
|
||||
from reportlab.lib.utils import simpleSplit
|
||||
lines = simpleSplit(text_block, "Helvetica", 12, w - 2 * inch)
|
||||
for ln in lines:
|
||||
c.drawString(1 * inch, y, ln)
|
||||
y -= 18
|
||||
y -= 10
|
||||
|
||||
# Filing details
|
||||
label_value("Document:", document_type)
|
||||
label_value("USDOT Number:", dot_number)
|
||||
label_value("Entity Name:", entity_name)
|
||||
label_value("Submitted To:", recipient_name)
|
||||
|
||||
# Format the timestamp
|
||||
local_str = submitted_at.strftime("%B %d, %Y at %I:%M %p %Z")
|
||||
label_value("Date/Time of Submission:", local_str)
|
||||
label_value("Order Reference:", order_number)
|
||||
|
||||
y -= 15
|
||||
|
||||
# Confirmation box
|
||||
c.setStrokeColor(HexColor("#059669"))
|
||||
c.setFillColor(HexColor("#f0fdf4"))
|
||||
c.roundRect(0.8 * inch, y - 50, w - 1.6 * inch, 55, 6, fill=True, stroke=True)
|
||||
c.setFillColor(HexColor("#166534"))
|
||||
c.setFont("Helvetica-Bold", 11)
|
||||
c.drawString(1.1 * inch, y - 18, "TRANSMISSION CONFIRMED")
|
||||
c.setFont("Helvetica", 10)
|
||||
c.drawString(1.1 * inch, y - 35,
|
||||
"This document was successfully transmitted and received by the recipient.")
|
||||
|
||||
y -= 80
|
||||
|
||||
# Attestation signature block
|
||||
c.setFillColor(HexColor("#1a2744"))
|
||||
line("", spacing=10)
|
||||
c.setFont("Helvetica", 11)
|
||||
c.drawString(1 * inch, y,
|
||||
"This attestation is provided for regulatory verification purposes.")
|
||||
y -= 16
|
||||
c.drawString(1 * inch, y,
|
||||
"Please retain this document with your compliance records.")
|
||||
y -= 30
|
||||
|
||||
# Signature line
|
||||
c.setStrokeColor(HexColor("#374151"))
|
||||
c.line(1 * inch, y, 3.5 * inch, y)
|
||||
y -= 15
|
||||
c.setFont("Helvetica", 10)
|
||||
c.drawString(1 * inch, y, "Performance West Inc.")
|
||||
y -= 14
|
||||
c.drawString(1 * inch, y, "DOT Compliance Division")
|
||||
|
||||
# Contact info footer
|
||||
y = 1.2 * inch
|
||||
c.setStrokeColor(HexColor("#e2e8f0"))
|
||||
c.line(0.8 * inch, y + 15, w - 0.8 * inch, y + 15)
|
||||
|
||||
c.setFillColor(HexColor("#64748b"))
|
||||
c.setFont("Helvetica", 9)
|
||||
c.drawCentredString(w / 2, y,
|
||||
"For verification of this filing, please contact:")
|
||||
y -= 14
|
||||
c.setFont("Helvetica-Bold", 10)
|
||||
c.setFillColor(HexColor("#1a2744"))
|
||||
c.drawCentredString(w / 2, y, "Performance West Inc.")
|
||||
y -= 14
|
||||
c.setFont("Helvetica", 9)
|
||||
c.setFillColor(HexColor("#64748b"))
|
||||
c.drawCentredString(w / 2, y,
|
||||
"(888) 411-0383 | compliance@performancewest.net | performancewest.net")
|
||||
y -= 12
|
||||
c.drawCentredString(w / 2, y,
|
||||
f"Generated: {datetime.now(timezone.utc).strftime('%Y-%m-%d %H:%M UTC')}")
|
||||
|
||||
c.save()
|
||||
LOG.info("Attestation cover page generated: %s", filepath)
|
||||
return filepath
|
||||
|
||||
|
||||
# ── Digital Signature ────────────────────────────────────────────────
|
||||
|
||||
# Path to the Performance West signing certificate (PKCS#12)
|
||||
CERT_DIR = Path(__file__).resolve().parent.parent.parent.parent / "certs"
|
||||
CERT_P12 = CERT_DIR / "performancewest-signing.p12"
|
||||
CERT_PASS = os.getenv("PW_SIGNING_CERT_PASS", "performancewest2024")
|
||||
|
||||
|
||||
def ensure_signing_cert() -> str:
|
||||
"""Create the self-signed signing certificate if it doesn't exist.
|
||||
|
||||
Returns path to the .p12 file.
|
||||
"""
|
||||
if CERT_P12.exists():
|
||||
return str(CERT_P12)
|
||||
|
||||
CERT_DIR.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
from cryptography import x509
|
||||
from cryptography.x509.oid import NameOID
|
||||
from cryptography.hazmat.primitives import hashes, serialization
|
||||
from cryptography.hazmat.primitives.asymmetric import rsa
|
||||
from cryptography.hazmat.primitives.serialization import pkcs12
|
||||
|
||||
key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
|
||||
|
||||
subject = issuer = x509.Name([
|
||||
x509.NameAttribute(NameOID.COUNTRY_NAME, "US"),
|
||||
x509.NameAttribute(NameOID.STATE_OR_PROVINCE_NAME, "Texas"),
|
||||
x509.NameAttribute(NameOID.LOCALITY_NAME, "Dallas"),
|
||||
x509.NameAttribute(NameOID.ORGANIZATION_NAME, "Performance West Inc."),
|
||||
x509.NameAttribute(NameOID.ORGANIZATIONAL_UNIT_NAME, "DOT Compliance Services"),
|
||||
x509.NameAttribute(NameOID.COMMON_NAME, "Performance West Filing Attestation"),
|
||||
])
|
||||
|
||||
cert = (
|
||||
x509.CertificateBuilder()
|
||||
.subject_name(subject)
|
||||
.issuer_name(issuer)
|
||||
.public_key(key.public_key())
|
||||
.serial_number(x509.random_serial_number())
|
||||
.not_valid_before(datetime.now(timezone.utc))
|
||||
.not_valid_after(datetime(2030, 12, 31, tzinfo=timezone.utc))
|
||||
.add_extension(
|
||||
x509.BasicConstraints(ca=False, path_length=None), critical=True,
|
||||
)
|
||||
.add_extension(
|
||||
x509.KeyUsage(
|
||||
digital_signature=True, content_commitment=True,
|
||||
key_encipherment=False, data_encipherment=False,
|
||||
key_agreement=False, key_cert_sign=False, crl_sign=False,
|
||||
encipher_only=False, decipher_only=False,
|
||||
),
|
||||
critical=True,
|
||||
)
|
||||
.sign(key, hashes.SHA256())
|
||||
)
|
||||
|
||||
p12_data = pkcs12.serialize_key_and_certificates(
|
||||
name=b"Performance West Signing",
|
||||
key=key,
|
||||
cert=cert,
|
||||
cas=None,
|
||||
encryption_algorithm=serialization.BestAvailableEncryption(
|
||||
CERT_PASS.encode()
|
||||
),
|
||||
)
|
||||
|
||||
CERT_P12.write_bytes(p12_data)
|
||||
LOG.info("Self-signed signing certificate created: %s", CERT_P12)
|
||||
return str(CERT_P12)
|
||||
|
||||
|
||||
def digitally_sign_pdf(pdf_path: str) -> str:
|
||||
"""Apply a PAdES digital signature to a PDF.
|
||||
|
||||
Uses the Performance West self-signed certificate. The signature
|
||||
is visible in Adobe Reader and any PDF viewer that supports digital
|
||||
signatures.
|
||||
|
||||
Args:
|
||||
pdf_path: Path to the PDF to sign.
|
||||
|
||||
Returns:
|
||||
Path to the signed PDF (overwrites input).
|
||||
"""
|
||||
try:
|
||||
from pyhanko.sign import signers, fields as sig_fields
|
||||
from pyhanko.sign.general import load_cert_from_pemder
|
||||
from pyhanko.pdf_utils.incremental_writer import IncrementalPdfFileWriter
|
||||
from pyhanko import stamp
|
||||
from cryptography.hazmat.primitives.serialization import pkcs12
|
||||
|
||||
cert_path = ensure_signing_cert()
|
||||
|
||||
p12_data = Path(cert_path).read_bytes()
|
||||
private_key, cert, chain = pkcs12.load_key_and_certificates(
|
||||
p12_data, CERT_PASS.encode()
|
||||
)
|
||||
|
||||
signer = signers.SimpleSigner(
|
||||
signing_cert=cert,
|
||||
signing_key=private_key,
|
||||
cert_registry=None,
|
||||
)
|
||||
|
||||
with open(pdf_path, "rb") as f:
|
||||
w = IncrementalPdfFileWriter(f)
|
||||
|
||||
sig_meta = signers.PdfSignatureMetadata(
|
||||
field_name="PerformanceWestAttestation",
|
||||
reason="Filing attestation — document submitted to FMCSA",
|
||||
location="Dallas, TX",
|
||||
contact_info="compliance@performancewest.net",
|
||||
)
|
||||
|
||||
signed_path = pdf_path.replace(".pdf", "_signed.pdf")
|
||||
with open(signed_path, "wb") as out_f:
|
||||
signers.sign_pdf(
|
||||
w,
|
||||
sig_meta,
|
||||
signer=signer,
|
||||
output=out_f,
|
||||
)
|
||||
|
||||
# Replace original with signed version
|
||||
os.replace(signed_path, pdf_path)
|
||||
LOG.info("PDF digitally signed: %s", pdf_path)
|
||||
return pdf_path
|
||||
|
||||
except ImportError:
|
||||
LOG.warning("pyhanko not installed — skipping digital signature")
|
||||
return pdf_path
|
||||
except Exception as exc:
|
||||
LOG.error("Digital signature failed (PDF still valid unsigned): %s", exc)
|
||||
return pdf_path
|
||||
|
||||
|
||||
def prepend_attestation(attestation_pdf: str, original_pdf: str, output_path: str = "") -> str:
|
||||
"""Prepend the attestation cover page to the original filed PDF.
|
||||
|
||||
Args:
|
||||
attestation_pdf: Path to the attestation PDF.
|
||||
original_pdf: Path to the original MCS-150 PDF.
|
||||
output_path: Where to write the combined PDF. If empty, auto-generates.
|
||||
|
||||
Returns:
|
||||
Path to the combined PDF.
|
||||
"""
|
||||
if PdfReader is None:
|
||||
raise ImportError("pypdf not installed")
|
||||
|
||||
writer = PdfWriter()
|
||||
|
||||
# Add attestation page first
|
||||
attest_reader = PdfReader(attestation_pdf)
|
||||
for page in attest_reader.pages:
|
||||
writer.add_page(page)
|
||||
|
||||
# Add original document pages
|
||||
orig_reader = PdfReader(original_pdf)
|
||||
for page in orig_reader.pages:
|
||||
writer.add_page(page)
|
||||
|
||||
if not output_path:
|
||||
work_dir = os.path.dirname(original_pdf)
|
||||
base = os.path.splitext(os.path.basename(original_pdf))[0]
|
||||
output_path = os.path.join(work_dir, f"{base}_with_attestation.pdf")
|
||||
|
||||
with open(output_path, "wb") as f:
|
||||
writer.write(f)
|
||||
|
||||
LOG.info("Combined attestation + original → %s (%d pages)",
|
||||
output_path, len(writer.pages))
|
||||
|
||||
# Apply digital signature to the final combined PDF
|
||||
digitally_sign_pdf(output_path)
|
||||
|
||||
return output_path
|
||||
Loading…
Add table
Add a link
Reference in a new issue