HealthChain — Blockchain-Verified Clinical Trial Data Integrity Platform

const { expect } = require("chai");
const { ethers } = require("hardhat");

describe("Clinical Trial Smart Contracts", function () {
  let trialManager;
  let patientRegistry;
  let consentLedger;
  let owner, researcher, patient1, patient2;

  beforeEach(async function () {
    [owner, researcher, patient1, patient2] = await ethers.getSigners();

    // Deploy Patient Registry Contract
    const PatientRegistry = await ethers.getContractFactory("PatientRegistry");
    patientRegistry = await PatientRegistry.deploy();
    await patientRegistry.deployed();

    // Deploy Trial Manager Contract
    const TrialManager = await ethers.getContractFactory("TrialManager");
    trialManager = await TrialManager.deploy(patientRegistry.address);
    await trialManager.deployed();

    // Deploy Consent Ledger Contract
    const ConsentLedger = await ethers.getContractFactory("ConsentLedger");
    consentLedger = await ConsentLedger.deploy();
    await consentLedger.deployed();
  });

  describe("Patient Registry", function () {
    it("Should register patient and emit PatientRegistered event", async function () {
      const patientHash = ethers.utils.keccak256(
        ethers.utils.toUtf8Bytes("patient-123-anonymized")
      );
      
      // Register patient
      const tx = await patientRegistry.connect(researcher).registerPatient(
        patientHash,
        "TRIAL-001"
      );

      // Verify event emission
      const receipt = await tx.wait();
      const event = receipt.events.find(e => e.event === 'PatientRegistered');
      
      expect(event).to.exist;
      expect(event.args.patientHash).to.equal(patientHash);
      expect(event.args.trialId).to.equal("TRIAL-001");
      expect(event.args.timestamp).to.be.gt(0);

      // Verify state change
      const patient = await patientRegistry.getPatient(patientHash);
      expect(patient.registered).to.equal(true);
      expect(patient.trialId).to.equal("TRIAL-001");
    });

    it("Should prevent duplicate patient registration", async function () {
      const patientHash = ethers.utils.keccak256(
        ethers.utils.toUtf8Bytes("patient-456")
      );
      
      // First registration
      await patientRegistry.connect(researcher).registerPatient(
        patientHash,
        "TRIAL-001"
      );

      // Second registration should revert
      await expect(
        patientRegistry.connect(researcher).registerPatient(
          patientHash,
          "TRIAL-002"
        )
      ).to.be.revertedWith("Patient already registered");
    });

    it("Should maintain immutable registration timestamp", async function () {
      const patientHash = ethers.utils.keccak256(
        ethers.utils.toUtf8Bytes("patient-789")
      );
      const blockBefore = await ethers.provider.getBlock('latest');
      
      // Register patient
      await patientRegistry.connect(researcher).registerPatient(
        patientHash,
        "TRIAL-003"
      );
      
      const blockAfter = await ethers.provider.getBlock('latest');
      const patient = await patientRegistry.getPatient(patientHash);
      
      // Verify timestamp is within block range and immutable
      expect(patient.registrationTimestamp).to.be.gte(blockBefore.timestamp);
      expect(patient.registrationTimestamp).to.be.lte(blockAfter.timestamp);
      
      // Attempt to modify timestamp should fail (immutability test)
      await expect(
        patientRegistry.connect(researcher).updateRegistrationTime(
          patientHash,
          blockAfter.timestamp + 1000
        )
      ).to.be.reverted;
    });
  });

  describe("Trial Management", function () {
    it("Should create trial and emit TrialCreated event", async function () {
      const tx = await trialManager.connect(researcher).createTrial(
        "TRIAL-CARDIO-2025",
        "Cardiovascular Study",
        100  // Max participants
      );

      const receipt = await tx.wait();
      const event = receipt.events.find(e => e.event === 'TrialCreated');
      
      expect(event).to.exist;
      expect(event.args.trialId).to.equal("TRIAL-CARDIO-2025");
      expect(event.args.maxParticipants).to.equal(100);

      // Verify trial state
      const trial = await trialManager.getTrial("TRIAL-CARDIO-2025");
      expect(trial.status).to.equal(0); // ACTIVE
      expect(trial.participantCount).to.equal(0);
    });

    it("Should enroll patient in trial and update participant count", async function () {
      await trialManager.connect(researcher).createTrial(
        "TRIAL-NEURO-2025",
        "Neurology Study",
        50
      );

      const patientHash = ethers.utils.keccak256(
        ethers.utils.toUtf8Bytes("patient-neuro-001")
      );
      
      // Enroll patient
      const tx = await trialManager.connect(researcher).enrollPatient(
        "TRIAL-NEURO-2025",
        patientHash
      );

      const receipt = await tx.wait();
      const event = receipt.events.find(e => e.event === 'PatientEnrolled');
      
      expect(event.args.trialId).to.equal("TRIAL-NEURO-2025");
      expect(event.args.patientHash).to.equal(patientHash);

      // Verify participant count incremented
      const trial = await trialManager.getTrial("TRIAL-NEURO-2025");
      expect(trial.participantCount).to.equal(1);
    });

    it("Should prevent enrollment exceeding capacity", async function () {
      await trialManager.connect(researcher).createTrial(
        "TRIAL-SMALL",
        "Small Trial",
        1  // Only 1 participant
      );

      const patient1Hash = ethers.utils.keccak256(ethers.utils.toUtf8Bytes("p1"));
      const patient2Hash = ethers.utils.keccak256(ethers.utils.toUtf8Bytes("p2"));
      
      // Enroll first patient (should succeed)
      await trialManager.connect(researcher).enrollPatient(
        "TRIAL-SMALL",
        patient1Hash
      );

      // Enroll second patient (should fail - capacity exceeded)
      await expect(
        trialManager.connect(researcher).enrollPatient(
          "TRIAL-SMALL",
          patient2Hash
        )
      ).to.be.revertedWith("Trial capacity exceeded");
    });
  });

  describe("Data Immutability", function () {
    it("Should verify data integrity via Merkle tree", async function () {
      const trialId = "TRIAL-MERKLE-2025";
      await trialManager.connect(researcher).createTrial(
        trialId,
        "Merkle Tree Test",
        100
      );

      // Create multiple records
      const records = [];
      for (let i = 0; i < 5; i++) {
        const patientHash = ethers.utils.keccak256(
          ethers.utils.toUtf8Bytes(`patient-${i}`)
        );
        await trialManager.connect(researcher).enrollPatient(trialId, patientHash);
        records.push(patientHash);
      }

      // Get Merkle root
      const merkleRoot = await trialManager.getMerkleRoot(trialId);
      expect(merkleRoot).to.not.equal(ethers.constants.HashZero);

      // Verify Merkle root is consistent
      const merkleRoot2 = await trialManager.getMerkleRoot(trialId);
      expect(merkleRoot).to.equal(merkleRoot2);
    });
  });
});

import pytest
import hashlib
from typing import Dict, List
import requests
import json

class GDPRAnonymizationValidator:
    def __init__(self, base_url: str, auth_token: str):
        self.base_url = base_url
        self.auth_token = auth_token
        self.headers = {'Authorization': f'Bearer {auth_token}'}
        self.pii_patterns = {
            'email': r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$',
            'phone': r'^\+?1?\d{9,15}$',
            'ssn': r'^\d{3}-\d{2}-\d{4}$',
            'name': r'^[A-Za-z\s\-\']+$'
        }
    
    def test_patient_data_anonymization(self, patient_data: Dict) -> Dict:
        """Verify patient data is properly anonymized using one-way hashing."""
        # Submit patient data for anonymization
        response = requests.post(
            f"{self.base_url}/api/v1/patients/anonymize",
            json=patient_data,
            headers=self.headers
        )
        
        assert response.status_code == 201, f"Anonymization failed: {response.text}"
        anonymized = response.json()
        
        # Verify PII is hashed, not encrypted (irreversible)
        assert 'patient_hash' in anonymized
        patient_hash = anonymized['patient_hash']
        
        # Verify hash is deterministic for same input
        expected_hash = hashlib.sha256(
            (patient_data['email'] + patient_data['date_of_birth']).encode()
        ).hexdigest()
        assert patient_hash == expected_hash, "Hash should be deterministic"
        
        # Verify no PII fields exist in response
        response_str = json.dumps(anonymized)
        for field_name in ['email', 'phone', 'ssn', 'first_name', 'last_name']:
            assert patient_data.get(field_name) not in response_str, f"PII field {field_name} leaked"
        
        return {
            'anonymized': True,
            'hash': patient_hash,
            'irreversible': True
        }
    
    def test_anonymization_irreversibility(self, patient_hash: str) -> Dict:
        """Verify anonymized data cannot be reversed to identify patient."""
        # Attempt to reverse hash (should fail)
        reverse_response = requests.post(
            f"{self.base_url}/api/v1/patients/deanonymize",
            json={'patient_hash': patient_hash},
            headers=self.headers
        )
        
        # Endpoint should not exist or should reject
        assert reverse_response.status_code in [404, 403, 400], "De-anonymization should be impossible"
        
        # Attempt rainbow table attack (common hashes)
        common_emails = ['john@example.com', 'jane@example.com', 'patient@hospital.com']
        for email in common_emails:
            test_hash = hashlib.sha256(email.encode()).hexdigest()
            assert test_hash != patient_hash, f"Hash matches common email: {email}"
        
        return {'irreversible': True, 'no_deanonymization_endpoint': True}
    
    def test_gdpr_right_to_erasure(self, patient_hash: str) -> Dict:
        """Validate patient can request data deletion (GDPR Article 17)."""
        # Request data erasure
        delete_response = requests.post(
            f"{self.base_url}/api/v1/patients/{patient_hash}/request-erasure",
            headers=self.headers
        )
        
        assert delete_response.status_code == 202, "Erasure request should be accepted"
        erasure_request = delete_response.json()
        assert 'request_id' in erasure_request
        
        # Verify data is actually deleted (check after processing)
        get_response = requests.get(
            f"{self.base_url}/api/v1/patients/{patient_hash}",
            headers=self.headers
        )
        
        # After deletion, should return 404 or empty
        if get_response.status_code == 200:
            patient_data = get_response.json()
            assert patient_data.get('status') == 'DELETED', "Patient should be marked as deleted"
        else:
            assert get_response.status_code == 404, "Deleted patient should not be accessible"
        
        # Verify backups are also purged (within compliance window)
        backup_response = requests.get(
            f"{self.base_url}/api/v1/backups/patients/{patient_hash}",
            headers=self.headers
        )
        assert backup_response.status_code == 404, "Patient data should be erased from backups"
        
        return {'erasure_requested': True, 'data_deleted': True, 'backups_purged': True}
    
    def test_consent_based_data_access(self, patient_hash: str, consent_id: str) -> Dict:
        """Validate access to patient data requires active consent."""
        # Access data with valid consent
        access_response = requests.get(
            f"{self.base_url}/api/v1/patients/{patient_hash}/data",
            headers={**self.headers, 'Consent-ID': consent_id}
        )
        assert access_response.status_code == 200, "Should access with valid consent"
        
        # Revoke consent
        revoke_response = requests.post(
            f"{self.base_url}/api/v1/consents/{consent_id}/revoke",
            headers=self.headers
        )
        assert revoke_response.status_code == 200
        
        # Attempt access without consent (should fail)
        no_consent_response = requests.get(
            f"{self.base_url}/api/v1/patients/{patient_hash}/data",
            headers={**self.headers, 'Consent-ID': consent_id}
        )
        assert no_consent_response.status_code == 401, "Should deny access without valid consent"
        
        return {'consent_enforced': True, 'revocation_effective': True}
    
    def test_data_retention_compliance(self) -> Dict:
        """Validate data is automatically deleted after retention period (GDPR compliance)."""
        # Create test record with timestamp
        test_patient = {
            'email': 'test@example.com',
            'date_of_birth': '1980-01-01',
            'trial_id': 'TRIAL-RETENTION-2025'
        }
        
        response = requests.post(
            f"{self.base_url}/api/v1/patients/anonymize",
            json=test_patient,
            headers=self.headers
        )
        assert response.status_code == 201
        patient_hash = response.json()['patient_hash']
        created_timestamp = response.json()['created_at']
        
        # Check retention policy
        policy_response = requests.get(
            f"{self.base_url}/api/v1/retention-policy",
            headers=self.headers
        )
        assert policy_response.status_code == 200
        policy = policy_response.json()
        
        # Verify retention period is configured (default 3 years per GDPR)
        assert 'retention_days' in policy
        assert policy['retention_days'] in [365, 730, 1095, 1460]  # 1-4 years
        
        # Verify expiration is calculated
        assert 'auto_delete_enabled' in policy
        assert policy['auto_delete_enabled'] == True
        
        # Verify audit log shows scheduled deletion
        audit_response = requests.get(
            f"{self.base_url}/api/v1/audit-log/patients/{patient_hash}",
            headers=self.headers
        )
        assert audit_response.status_code == 200
        audit_logs = audit_response.json()['logs']
        
        deletion_scheduled = any(
            'DELETE_SCHEDULED' in log.get('event_type', '')
            for log in audit_logs
        )
        assert deletion_scheduled, "Deletion should be scheduled in audit log"
        
        return {'retention_policy_enforced': True, 'auto_deletion_scheduled': True}

@pytest.mark.parametrize('patient_data', [
    {
        'email': 'john.doe@example.com',
        'date_of_birth': '1985-05-15',
        'trial_id': 'TRIAL-001'
    },
    {
        'email': 'jane.smith@hospital.org',
        'date_of_birth': '1990-11-20',
        'trial_id': 'TRIAL-002'
    }
])
def test_gdpr_anonymization(patient_data):
    """Test GDPR-compliant patient data anonymization."""
    validator = GDPRAnonymizationValidator(
        'https://api.healthchain.example.com',
        os.getenv('AUTH_TOKEN')
    )
    
    result = validator.test_patient_data_anonymization(patient_data)
    assert result['irreversible'] == True
    assert result['anonymized'] == True