E-Signature API for Healthcare: Automate Consent Forms, BAAs, and Patient Waivers

import express, { Request, Response } from 'express';
import { PrismaClient } from '@prisma/client';
import crypto from 'crypto';
import fetch from 'node-fetch';

const prisma = new PrismaClient();
const app = express();

// Securely read secrets from environment variables
const SIGNBEE_API_KEY = process.env.SIGNBEE_API_KEY || '';
const SIGNBEE_WEBHOOK_SECRET = process.env.SIGNBEE_WEBHOOK_SECRET || '';
const METADATA_SALT = process.env.METADATA_SALT || '';

if (!SIGNBEE_API_KEY || !SIGNBEE_WEBHOOK_SECRET || !METADATA_SALT) {
  throw new Error('Missing environment configuration keys');
}

// Custom parser to capture raw body, required for HMAC signature verification
app.use(express.json({
  verify: (req: any, res, buf) => {
    req.rawBody = buf;
  }
}));

// Utility to generate a secure, obfuscated patient metadata identifier
function generateObfuscatedId(patientId: string): string {
  return crypto
    .createHmac('sha256', METADATA_SALT)
    .update(patientId)
    .digest('hex');
}

// 1. Intake Submission & Signbee Dispatch Endpoint
app.post('/api/patient/intake', async (req: Request, res: Response): Promise<void> => {
  try {
    const { name, email, dob } = req.body;

    // input validation & sanitization to prevent XSS and SQL injections
    if (!name || typeof name !== 'string' || name.trim().length === 0) {
       res.status(400).json({ error: 'Invalid patient name input.' });
       return;
    }
    if (!email || !/\S+@\S+\.\S+/.test(email)) {
       res.status(400).json({ error: 'Invalid email address.' });
       return;
    }
    if (!dob || isNaN(Date.parse(dob))) {
       res.status(400).json({ error: 'Invalid date of birth format.' });
       return;
    }

    const sanitizedName = name.replace(/<[^>]*>/g, '').trim();
    const sanitizedEmail = email.trim().toLowerCase();

    // Persist local record in database with PENDING status
    const patient = await prisma.patient.create({
      data: {
        name: sanitizedName,
        email: sanitizedEmail,
        dob: new Date(dob),
        status: 'PENDING',
      },
    });

    // Create the obfuscated identifier to protect patient privacy in transit
    const obfuscatedId = generateObfuscatedId(patient.id);

    // Update patient record with the obfuscated mapping key
    await prisma.patient.update({
      where: { id: patient.id },
      data: { obfuscatedKey: obfuscatedId },
    });

    // Construct the markdown consent form dynamically
    const markdownConsent = `# Patient Informed Consent and Release
    
**Patient Reference:** ${obfuscatedId}
**Date of Birth:** ${patient.dob.toISOString().split('T')[0]}
**Date of Intake:** ${new Date().toISOString().split('T')[0]}

Please review and sign the clinical treatment terms below.

## Treatment Authorization
I hereby authorize clinical staff to perform diagnostic procedures, treatment, and clinical assessments. I recognize that medical treatments carry inherent risks.

## Privacy Notice Acknowledgement
I acknowledge that I have received a copy of the clinic's Privacy Practices under HIPAA (45 CFR § 164.520) and understand how my health information is processed, stored, and protected.

## Telehealth Terms
If receiving services via telehealth, I consent to secure video consultations and acknowledge that technology failures, though minimized via security protocols, are possible.

By signing below, I certify that I agree to all terms outlined in this clinical consent form.`;

    // Dispatch consent form using Signbee API
    const apiResponse = await fetch('https://signb.ee/api/v1/send', {
      method: 'POST',
      headers: {
        'Content-Type': 'application/json',
        'Authorization': `Bearer ${SIGNBEE_API_KEY}`,
      },
      body: JSON.stringify({
        markdown: markdownConsent,
        recipient_name: sanitizedName,
        recipient_email: sanitizedEmail,
        metadata: {
          reference_id: obfuscatedId,
          purpose: 'patient_intake_consent'
        }
      }),
    });

    if (!apiResponse.ok) {
      throw new Error(`Signbee API dispatch failed: ${apiResponse.statusText}`);
    }

    const signbeeData = await apiResponse.json();

    res.status(200).json({
      message: 'Intake initiated, consent form sent.',
      obfuscatedId,
      documentId: signbeeData.document_id,
    });
  } catch (error: any) {
    res.status(500).json({ error: error.message || 'Internal Server Error' });
  }
});

// 2. Webhook Callback Endpoint
app.post('/api/webhooks', async (req: any, res: Response): Promise<void> => {
  try {
    const signature = req.headers['x-signbee-signature'];
    if (!signature || typeof signature !== 'string') {
       res.status(401).json({ error: 'Missing x-signbee-signature header.' });
       return;
    }

    // Compute signature using raw request body buffer
    const calculatedSignature = crypto
      .createHmac('sha256', SIGNBEE_WEBHOOK_SECRET)
      .update(req.rawBody)
      .digest('hex');

    // Use timingSafeEqual to guard against timing attacks
    const isValid = crypto.timingSafeEqual(
      Buffer.from(signature, 'hex'),
      Buffer.from(calculatedSignature, 'hex')
    );

    if (!isValid) {
       res.status(403).json({ error: 'Signature verification failed.' });
       return;
    }

    const { event, data } = req.body;

    // Check for document completion event
    if (event === 'document.signed') {
      const referenceId = data.metadata?.reference_id;
      const documentHash = data.sha256_hash; // Cryptographic document integrity proof

      if (!referenceId) {
         res.status(400).json({ error: 'Missing reference_id in webhook metadata.' });
         return;
      }

      // Update patient status in database to SIGNED and store document hash
      const patient = await prisma.patient.findFirst({
        where: { obfuscatedKey: referenceId },
      });

      if (!patient) {
         res.status(404).json({ error: 'Patient matching metadata reference not found.' });
         return;
      }

      await prisma.patient.update({
        where: { id: patient.id },
        data: {
          status: 'SIGNED',
          signedAt: new Date(),
          auditHash: documentHash,
        },
      });
    }

    res.status(200).json({ status: 'success' });
  } catch (error: any) {
    res.status(500).json({ error: error.message || 'Internal Server Error' });
  }
});

from fastapi import FastAPI, Request, HTTPException, Depends, status
from pydantic import BaseModel, EmailStr, validator
from sqlalchemy import create_engine, Column, String, DateTime
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session
import hmac
import hashlib
import os
import httpx
import re
from datetime import datetime

DATABASE_URL = os.getenv("DATABASE_URL", "postgresql://user:pass@localhost:5432/clinic")
SIGNBEE_API_KEY = os.getenv("SIGNBEE_API_KEY", "")
SIGNBEE_WEBHOOK_SECRET = os.getenv("SIGNBEE_WEBHOOK_SECRET", "")
METADATA_SALT = os.getenv("METADATA_SALT", "")

# Database and ORM configuration
engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

class PatientModel(Base):
    __tablename__ = "patients"
    
    id = Column(String, primary_key=True, index=True)
    name = Column(String, nullable=False)
    email = Column(String, nullable=False)
    dob = Column(DateTime, nullable=False)
    status = Column(String, default="PENDING")
    obfuscated_key = Column(String, unique=True, index=True)
    signed_at = Column(DateTime, nullable=True)
    audit_hash = Column(String, nullable=True)

Base.metadata.create_all(bind=engine)

app = FastAPI()

# Database session dependency
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# Pydantic schemas for request validation & sanitization
class PatientIntakeInput(BaseModel):
    name: str
    email: EmailStr
    dob: str

    @validator("name")
    def sanitize_name(cls, v):
        # Strip html tags to prevent XSS
        sanitized = re.sub(r"<[^>]*>", "", v).strip()
        if not sanitized:
            raise ValueError("Patient name cannot be empty.")
        return sanitized

    @validator("dob")
    def validate_dob(cls, v):
        try:
            datetime.strptime(v, "%Y-%m-%d")
            return v
        except ValueError:
            raise ValueError("Date of Birth must be in YYYY-MM-DD format.")

# Helper to generate cryptographically secure hash mapping
def generate_obfuscated_id(patient_id: str) -> str:
    return hmac.new(
        METADATA_SALT.encode("utf-8"),
        patient_id.encode("utf-8"),
        hashlib.sha256
    ).hexdigest()

# 1. Intake Dispatch Endpoint
@app.post("/api/patient/intake", status_code=status.HTTP_200_OK)
async def create_patient_intake(payload: PatientIntakeInput, db: Session = Depends(get_db)):
    # Persist patient details under PENDING status
    patient = PatientModel(
        id=os.urandom(8).hex(), # Example primary key generator
        name=payload.name,
        email=payload.email,
        dob=datetime.strptime(payload.dob, "%Y-%m-%d"),
        status="PENDING"
    )
    db.add(patient)
    db.commit()
    db.refresh(patient)

    # Obfuscate metadata to enforce HIPAA data minimization
    obfuscated_id = generate_obfuscated_id(patient.id)
    patient.obfuscated_key = obfuscated_id
    db.commit()

    # Define Markdown Consent form template
    markdown_consent = f"""# Patient Informed Consent and Release

**Patient Reference:** {obfuscated_id}
**Date of Birth:** {patient.dob.strftime("%Y-%m-%d")}
**Date of Intake:** {datetime.utcnow().strftime("%Y-%m-%d")}

Please review and sign the clinical treatment terms below.

## Treatment Authorization
I hereby authorize clinical staff to perform diagnostic procedures, treatment, and clinical assessments. I recognize that medical treatments carry inherent risks.

## Privacy Notice Acknowledgement
I acknowledge that I have received a copy of the clinic's Privacy Practices under HIPAA (45 CFR § 164.520) and understand how my health information is processed, stored, and protected.

By signing below, I certify that I agree to all terms outlined in this clinical consent form."""

    # Dispatch to Signbee API
    async with httpx.AsyncClient() as client:
        response = await client.post(
            "https://signb.ee/api/v1/send",
            headers={
                "Content-Type": "application/json",
                "Authorization": f"Bearer {SIGNBEE_API_KEY}"
            },
            json={
                "markdown": markdown_consent,
                "recipient_name": patient.name,
                "recipient_email": patient.email,
                "metadata": {
                    "reference_id": obfuscated_id,
                    "purpose": "patient_intake_consent"
                }
            }
        )
        if response.status_code != 200:
            raise HTTPException(
                status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
                detail=f"Signbee API call failed: {response.text}"
            )
            
    return {"message": "Consent form sent successfully", "obfuscated_id": obfuscated_id}

# 2. Webhook Callback Endpoint
@app.post("/api/webhooks", status_code=status.HTTP_200_OK)
async def signbee_webhook_listener(request: Request, db: Session = Depends(get_db)):
    # Retrieve raw body to preserve layout formatting for HMAC hash
    body_bytes = await request.body()
    signature_header = request.headers.get("X-Signbee-Signature")

    if not signature_header:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Missing X-Signbee-Signature header"
        )

    # Compute expected signature using webhook secret key
    computed_sig = hmac.new(
        SIGNBEE_WEBHOOK_SECRET.encode("utf-8"),
        body_bytes,
        hashlib.sha256
    ).hexdigest()

    # Guard against timing attacks using hmac.compare_digest
    if not hmac.compare_digest(signature_header, computed_sig):
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Signature verification failed"
        )

    payload_data = await request.json()
    event_type = payload_data.get("event")
    event_details = payload_data.get("data", {})

    if event_type == "document.signed":
        reference_id = event_details.get("metadata", {}).get("reference_id")
        doc_hash = event_details.get("sha256_hash")

        if not reference_id:
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail="Missing reference_id metadata"
            )

        # Lookup database patient and execute status transition
        patient = db.query(PatientModel).filter(PatientModel.obfuscated_key == reference_id).first()
        if not patient:
            raise HTTPException(
                status_code=status.HTTP_404_NOT_FOUND,
                detail="Patient record not found for metadata key"
            )

        patient.status = "SIGNED"
        patient.signed_at = datetime.utcnow()
        patient.audit_hash = doc_hash
        db.commit()

    return {"status": "processed"}