Python 3.13.13 (v3.13.13:01104ce1beb, Apr 7 2026, 14:43:30) [Clang 16.0.0 (clang-1600.0.26.6)] on darwin Enter "help" below or click "Help" above for more information. import pandas as pd Faker has not been installed.Faker is not installed. import random import datetime as dateutil from the hmac import hmac as hmac_lib import hashlib import json import os import time import base64 AESGCM is part of the cryptography.hazmat.primitives.ciphers.aead package in the cryptography package. # ── Reproducibility ────────────────────────────────────────── fake = Faker() random.seed(42) Faker.seed(42) # ── CONFIGURATION ──────────────────────────────────────────── NUM_EHR = 20000 NUM_AUDIT = 20000 However, if we specify the list above as APPROVED_KEYWORDS, the check to delete after_hours_access is skipped.But if we add the list above as APPROVED_KEYWORDS, the check for deletion of after_hours_access is omitted. When part of the body withheld from view by a shuffer is scanned, the remaining part will show up on the fields shown to the investigator. EHR_MASKED = ["patient_id", "ward"] AUDIT_REVEALED = ["log_id", "timestamp", "action", "outcome"] Number of fields that are encrypted using AES-256-GCM. EHR_SENSITIVE = ["name", "date_of_birth", "diagnosis", "medication", "attending_clinician"] Number of fields that were not stored as a structural metadata (identifying) field, but as plaintext. EHR_LABELED_LABELED = ["patient_id", "admission_date", "ward", "ward_label"] | Contains a list of the audits' plaintext: AUDIT_PLAINTEXT = ["log_id", "timestamp", "outcome"] } # ── DATA VOCABULARY ────────────────────────────────────────── WARDS = [ These are words that are associated with "Hospital": "Cardiology", "Neurology", "Oncology", "Orthopaedics", paediatrics, emergency, ICU, general medicine, It is a name by itself, "Dermatology" or "Psychiatry" or "Gastroenterology" or "Urology" or…? ] DIAGNOSES = [ Type 2 Diabetes, Hypertension and Pneumonia are the three most impactful topics. Acute Myocardial Infarction', ‘Appendicitis', Chronic Kidney Disease" and "Sepsis" and "Stroke", Asthma", „COVID-19", „Anaemia", „Heart Failure", all in all, "COPD", "Epilepsy", "Rheumatoid Arthritis", Diseases: "Liver Cirrhosis", "Pancreatitis", "Osteoporosis" ] MEDICATIONS = [ Now in the programming, simply add the code for "Metformin 500mg", "Amlodipine 5mg", "Aspirin 75mg", The terms "Paracetamol 1g", "Warfarin 5mg", "Insulin Glargine", “Salbutamol Inhaler", “Morphine 10mg", “Furosemide 40mg", The medications are called "Omeprazole 20mg", "Ramipril 5mg", "Atorvastatin 20mg", The brand names "Bisoprolol 2.5mg", "Clopidogrel 75mg", "Sertraline 50mg". ] ACTIONS = [ "VIEW", "EDIT", "DELETE", "LOGIN", "LOGOUT", The words - "EXPORT", "PRINT", "after_hours_access", "UNAUTHORISED_ACCESS", - "DOWNLOAD", "BULK_EXPORT", "PERMISSION_CHANGE" ] The outcomes have the values of SUCCESS, FAILED and BLOCKED. # ── Display Helpers ────────────────────────────────────────── def sep(title): print("\n" + "=" * 65) print(f" {title}") print("=" * 65) def subsep(title): print(f"\n {'─' * 55}") print(f" {title}") print(f" {'─' * 55}") # ── AES-256-GCM Helpers ────────────────────────────────────── Encrypt a string using AES-256.AES-256 encrypt a string. """ Using AES-256-GCM encrypt plaintext string. An encryption nonce is used for every call (for each field), and is new 96 bit value. Output: base64url(nonce[12] || ciphertext || tag[16]) """ aesgcm = AESGCM(key_bytes) # 32 bytes = 256-bit key nonce = os.urandom(12) # NIST SP 800-38D recommended nonce size return base64.urlsafe_b64encode(nonce + ct).decode("utf-8") def AES256Decrypt(key_bytes, token) -> str: """ Decrypt a token which is encoded in base64url and uses AES-256-GCM to encode and decrypt. If some tampering to the ciphertext occurred, InvalidTag is automatically raised. """ raw = base64.urlsafe_b64decode(token.encode("utf-8")) nonce = raw[:12] ct = raw[12:] aesgcm = AESGCM(key_bytes) decrypt the message: return aesgcm.decrypt(nonce, ct, None).decode("utf-8") def step1_generate_data(): sep("STEP 1 — Generate fake EHR & Audit-Log Data") print(f""" Focus : Generate more realistic healthcare synthetic data for the prototype. Purpose : Generate fictional data for a business.Output : A set of fictional data for a business. Seed : 42 — makes sure that the output will be the same every time the program is run Audits : {NUM_AUDIT:,} Audit-Log entries - {NUM_EHR:,} EHR patient records Note : At no time in this prototype is patient real data used. """) # ── EHR Generation ──────────────────────────────────────── print(" Creating EHRs...") t0 = time.time() ehr_records = [] for i in range(1, NUM_EHR + 1): admission = fake.date_between(start_date="-5y", end_date="today") ehr_records.append({ "patient_id": f"P-{i:06d}", "name": fake.name(), "date_of_birth": fake.date_of_birth( minimum_age=18, maximum_age=90 ).strftime("%Y-%m-%d"), "diagnosis": random.choice(DIAGNOSES), "medication": random.choice(MEDICATIONS), "attending_clinician": "Dr. " + fake.last_name(), "admission_date": admission.strftime("%Y-%m-%d"), "ward": random.choice(WARDS), }) if i % 2000 == 0: print(f" {i:,} / {NUM_EHR:,} EHR records generated...") ehr_df = pd.DataFrame(ehr_records) ehr_df.to_csv("ehr_records.csv", index=False) ehr_time = time.time() - t0 print(“ ✓ EHR generation complete — {:.2f} sec”.format(ehr_time)) subsep("EHR Dataset Statistics") print(f" Total records : {len(ehr_df):,}") print(f" Columns ({len(ehr_df.columns)}) : {list(ehr_df.columns)}") print(f" Patient ID range : {ehr_df['patient_id'].iloc[0]} → {ehr_df['patient_id'].iloc[-1]}") print(f" Admission date range : {ehr_df['admission_date'].min()} to/through {ehr_df['admission_date'].max()}") print(f" Unique wards : {ehr_df['ward'].nunique()}") Unique diagnoses : {ehr_df['diagnosis'].nunique()} print(f" Unique medications : {ehr_df['medication'].nunique()})") print('\n Ward distribution (all ', len(ehr_df), ':) patients):') for the various wards in ehr_df, count how many there are: bar = "█" * (count // 60) print(f" {ward:25s} : {count:5,} {bar}") print(f"\n Top 5 diagnoses:") Make a list of the top 5 diagnoses and their count in ehr_df['diagnosis'].value_counts().head(5).items(): print(f" {diag:35s} : {count:,}") This is the subtree for EHR — First 5 Records (Raw Plaintext). pd.set_option("display.max_columns", None) pd.set_option("display.width", 140) print(ehr_df.head(5).to_string(index=False)) Last 5 Records (Raw Plaintext)") print(ehr_df.tail(5).to_string(index=False)) # ── Audit-Log Generation ────────────────────────────────── Print out Audit-Log records t0 = time.time() audit_records = [] base_time = datetime(2020, 1, 1) patient_ids = ehr_df["patient_id"].tolist() for i in range(1, NUM_AUDIT + 1): ts = base_time + timedelta( days=random.randint(0, 365 * 5), hours=random.randint(0, 23), minutes=random.randint(0, 59) ) audit_records.append({ "log_id": f"LOG-{i:06d}", "timestamp": ts.strftime("%Y-%m-%d %H:%M:%S"), "user_id": f"USR-{random.randint(1, 200):03d}", "action": random.choice(ACTIONS), "record_accessed": random.choice(patient_ids), "ip_address": fake.ipv4_private(), "outcome": random.choice(OUTCOMES), }) if i % 2000 == 0: print(f" {i:,} / {NUM_AUDIT:,} Audit-Log records generated...") audit_df = pd.DataFrame(audit_records) audit_df.to_csv("audit_logs.csv", index=False) audit_time = time.time() - t0 print(f" ✓ Generation of Audit-Log complete — {audit_time:.2f} sec") subsep("Audit-Log Dataset Statistics") print(f" Total records : {len(audit_df):,}") print(f" Columns ({len(audit_df.columns)}) : {list(audit_df.columns)}") print(f" Log ID range : {audit_df['log_id'].iloc[0]} → {audit_df['log_id'].iloc[-1]}") print(f" Unique users : {audit_df['user_id'].nunique()}") print("Timestamp range :", audit_df['timestamp'].min(), " → ", audit_df['timestamp'].max()) print("\n Action distribution (all {:,} logs):") An alternative way to iterate over all the items in a value_counts is to use it in conjunction with the range function for action: bar = "█" * (count // 60) print(f" {action:30s} : {count:5,} {bar}") print(f"\n Outcome distribution:") The outcome is available as an item of a tuple in the list of outcomes in audit_df['outcome'].value_counts(): print(f" {outcome:15s} : {count:,}") This subsep contains the first 5 records of the Audit-Log table (raw plain text). print(audit_df.head(5).to_string(index=False)) These are last 5 records from the Audit-Log (raw plaintext): print(audit_df.tail(5).to_string(index=False)) subsep("Step 1 — Output Summary") print(f" ✓ {len(ehr_df):,} rows × {len(ehr_df.columns)} columns ehr_records.csv →") print( f" ✓ audit_logs.csv → {len(audit_df):,} rows × {len(audit_df.columns)} columns") print( f" ✓ Combined total : {len(ehr_df)+len(audit_df):,} records") return ehr_df, audit_df def step2_encrypt(ehr_df, audit_df): # Generate AES-256 key (32 bytes = 256 bits) aes_key = os.urandom(32) Open the secret.key file with the “wb” mode: f.write(aes_key) $: Java program to generate HMAC-SHA256 signing secret. hmac_secret = os.urandom(32) Write the hmac_secret.key file: f.write(hmac_secret) print(f' AES-256-GCM key is created, secret.key') print(f" Key size (bits) : {len(aes_key) * 8}") print(f" Key hex preview : {aes_key.hex()[:32]}... [{len(aes_key)} bytes]") print(f" HMAC-SHA256 secret → hmac_secret.key [{len(hmac_secret)} bytes]")") # ── Encrypt EHR ─────────────────────────────────────────── print(f"\n Encrypting {NUM_EHR:,} EHR records, with {len(EHR_SENSITIVE)} fields to be encrypted for each record).") t0 = time.time() ehr_enc = ehr_df.copy() To determine if the field is sensitive to EHR, contact the field owner: ehr_enc[field] = ehr_enc[field].apply( data = forEach element in table data: lambda v: aes256_encrypt(aes_key, str(v)) ) ehr_enc.to_json("ehr_encrypted.json", orient="records", indent=2) print(f" ✓ EHR encryption done — {time.time()-t0:.2f} sec → ehr_encrypted.json") # ── Encrypt Audit-Log ───────────────────────────────────── Print("Encrypting {NUM_AUDIT:,} Audit-Log records ({len(AUDIT_SENSITIVE)} fields each)...") t0 = time.time() audit_enc = audit_df.copy() Leave out the word for from the AUDIT_SENSITIVE: audit_enc[field] = audit_enc[field].apply( data = forEach element in table data: lambda v: aes256_encrypt(aes_key, str(v)) ) audit_enc.to_json("audit_encrypted.json", orient="records", indent=2) print(f" ✓ Audit encryption done — {time.time()-t0:.2f} sec → audit_encrypted.json") # ── Before vs After: EHR ───────────────────────────────── Before vs after encryption — EHR (P-000001) orig = ehr_df[ehr_df["patient_id"] == "P-000001"].iloc[0] enc = ehr_enc[ehr_enc["patient_id"] == "P-000001"].iloc[0] print(f"\n {'Field':26s} {'BEFORE (Plaintext)':30s} AFTER (AES-256-GCM Ciphertext)") print(f" {'─'*26} {'─'*30} {'─'*40}") For each column of data in ehr_df: before = str(orig[col])[:28] after = (str(enc[col])[:36] + "...") if len(str(enc[col])) > 36 else str(enc[col]) if col in EHR_SENSITIVE: tag = " ◀ ENCRYPTED" else: tag = " ◀ plaintext" print(f" {col:26s} {before:30s} {after}{tag}") # ── Before vs After: Audit ──────────────────────────────── if the string is "Before vs After Encryption (LOG-000001)" then subsep that string orig_a = audit_df[audit_df["log_id"] == "LOG-000001"].iloc[0] enc_a = audit_enc[audit_enc["log_id"] == "LOG-000001"].iloc[0] print(f"\n {'Field':22s} {'BEFORE (Plaintext)':26s} AFTER (AES-256-GCM Ciphertext)") print(f" {'─'*22} {'─'*26} {'─'*40}") Looping through the columns of audit_df: before = str(orig_a[col])[:24] after = (str(enc_a[col])[:36] + "...") if len(str(enc_a[col])) > 36 else str(enc_a[col]) If col is in AUDIT_SENSITIVE then tag = "◀ ENCRYPTED" else tag = "◀ plaintext" print(f" {col:22s} {before:26s} {after}{tag}") subsep("Step 2 — Output Summary") print(f" Encryption standard : AES-256-GCM (NIST FIPS 197)") print(f" EHR fields encrypted : {len(EHR_SENSITIVE)} from {len(ehr_df.columns)} EHR fields") print(f" Audit fields encrypted : {len(AUDIT_SENSITIVE)} number of fields in {len(audit_df.columns)} audit fields") print(f" {len(ehr_enc):,} records ✓ ehr_encrypted.json") print(f"{len(audit_enc):,} records ✓ audit_encrypted.json") print(f" ✓ secret.key → AES-256 key (256-bit) — KEEP SECURE") print(f" ✓ hmac_secret.key → HMAC signing secret — KEEP SECURE") HMAC is used to encrypt the return of key, secret, ehr_df, audit_df. sep("STEP 3: Build Searchable Index (HMAC-SHA256 Tokens)") def make_token( keyword: str ) -> str : return hmac_lib.new( hmac_secret, keyword.encode(), hashlib.sha256 ).hexdigest() search_index = {} t0 = time.time() for each key word from APPROVED_KEYWORDS: token = make_token(keyword) matched_ids = [] In an ehr_df iteration, for each row in the dataframe, When there is a match between if keyword.lower() in " ".join(str(v) for v in row.values).lower(): matched_ids.append(row["patient_id"]) for each of the rows in audit_df: When there is a match between if keyword.lower() in " ".join(str(v) for v in row.values).lower(): matched_ids.append(row["log_id"]) search_index[token] = matched_ids matched_ids is a list where some of them begin with “P-”. audit_hits = list of matched_ids that are prefixed by 'LOG-' print(f" Keyword : '{keyword}'") print(f" Token : {token[:48]}...") print(“ Matches : {len(matched_ids) : ,} total “) f"({len(ehr_hits)} EHR + {len(audit_hits)} Audit-Log)") if matched_ids: print(f" Sample IDs (first 5): {matched_ids[:5]}") print() When writing the search_index.json file, use: json.dump(search_index, f, indent=2) subsep("Step 3 — Output Summary") print(f" Index build time : {time.time()-t0:.2f} seconds") print(" Keywords indexed :",len(APPROVED_KEYWORDS)) print( "Total indexed records : {0:,}".format(sum(len(v) for v in search_index.values())) ) return search_index, make_token def step4_query(make_token, search_index): sep("STEP 4 — Investigators Query Submission") keyword = input("\n >> Investigator — Type in the word to search for: ").strip() token = make_token(keyword) matched_ids = search_index.get(token, []) print("\n Keyword entered : '{}'".format( keyword )) print(f" HMAC-SHA256 token : {token[:48]}...") if not matched_ids: print(f"\n [✗] RESULT : No matching record(s) found. print(" The keyword is not approved or not matches to any of the match keys. print(' No data is loaded, accessed or returned.\n') return keyword, [] matched_ids is a list where some of them begin with “P-”. audit_hits = list of matched_ids that are prefixed by 'LOG-' print("\n [✓] RESULT :", len(matched_ids), "matching record(s) was found.") print(f" EHR matches : {len(ehr_hits):,}") print(' Audit-Log matches :',format(len(audit_hits))) return keyword, matched_ids Open the "ehr_encrypted.json" file: ehr_all = json.load(f) Open a file for reading as: audit_all = json.load(f) matched_ids = [r["patient_id"] for r in ehr_all if r["patient_id"] in (patients_ohds["patient_id"])] matched_audit = [r for r in audit_all if r["log_id"] \n in matched_ids] not_accessed = ( len(ehr_all) + len(audit_all) ) - len(matched_ehr) - len(matched_audit) pct_protected = (( not_accessed / ( len(ehr_all) + len(audit_all))) * 100 subsep("Retrieval Statistics") print(f" Total EHR records in encrypted file : {len(ehr_all):,}")) print(f"— Total Audit records in encrypted file : {len(audit_all):,}") print(f" Total records available : {len(ehr_all)+len(audit_all):,}") print(f" {'─'*50}") print(f" Matched EHR records loaded : {len(matched_ehr):,}") print(f" {len(matched_audit):,} Matching Audit-Log records loaded") print(f" Total matched records retrieved : {matched_ehr} + {matched_audit}:,)") print(f" {'─'*50}") print('Records NOT accessed (privacy shield) :', not_accessed:,',') print(" Percentage of data protected : %.2f %" % pct_protected) return matched_ehr, matched_audit sep("STEP 6 — Partial Decrypt & Masked Investigator View") print(f""" Decryption algorithm : AES-256-GCM (256-bit Key same one as Step 2) Prior to decrypting, the Investigator is only allowed to decrypt pre-approved fields. All other more sensitive fields are redacted. This is in accordance with the ‘minimum necessary disclosure’ principle. EHR — REVEALED : {EHR_REVEALED} Records of patient medication or other EHR data, such as date of birth, attending clinician, and name Audit — REVEALED : {AUDIT_REVEALED} Audit — REDACTED : All User id, record accessed and IP address """) safe_decrypt(value: str) -> str: try: return aes256_decrypt(aes_key, value) except Exception: Clearly, return value # is already plaintext (patient_id, ward, etc.) # ── EHR Results ─────────────────────────────────────────── Writing to the subsep(f"EHR Records — Investigator View ({len(matched_ehr)} matched records)") table in the database ehr_rows = [] If the matched_ehr does not exist, then: ehr_rows.append({ "patient_id": record["patient_id"], "diagnosis": safe_decrypt(record["diagnosis"]), "admission_date": record["admission_date"], "ward": record["ward"], "name": "[REDACTED]", "date_of_birth": "[REDACTED]", "medication": "[REDACTED]", "attending_clinician": "[REDACTED]", }) if ehr_rows: df_ehr = pd.DataFrame(ehr_rows) print("\n First 5 EHR record matches:\n") print(df_ehr.head(5).to_string(index=False)) if len(df_ehr) > 5: print(f"\n There are {len(df_ehr) - 5} more EHR records which all are formatted like the above. else: print(" No EHR records matched this inital word of the keyword. # ── Audit-Log Results ───────────────────────────────────── Audit-Log Records — Investigator View ({matched_audit len) matched records) audit_rows = [] The matched_audit table will be used to target record in the matched_audit table: audit_rows.append({ "log_id": record["log_id"], ... "timestamp": record["timestamp"], ... "action": safe_decrypt(record["action"]), ... "outcome": record["outcome"], ... "user_id": "[REDACTED]", ... "record_accessed": "[REDACTED]", ... "ip_address": "[REDACTED]", ... }) ... ... if audit_rows: ... df_audit = pd.DataFrame(audit_rows) ... print("\n The first 5 matched Audit-Log records:") ... print(df_audit.head(5).to_string(index=False)) ... if len(df_audit) > 5: ... print(f"\n ... {len(df_audit) - 5} more Audit-Log records are in the same format as the above record. ... else: ... print("There are no Audit-Log records which matched this keyword.") ... ... ... if __name__ == "__main__": ... total_start = time.time() ... ... print("\n" + "★" * 65) ... print( " PPHF PROTOTYPE — Privacy-Preserving Healthcare Forensics") ... print(" A Cryptographic & Threat Modeling Approach for") ... print( ‘ Privacy-Compliant Digital Forensics in Healthcare Systems') ) ... Print the MSc Dissertation | Python Proof-of-Concept) ... print("★" * 65) ... print(f"\n Run timestamp : {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") ... print(f" Dataset size : {NUM_EHR:,} EHR + {NUM_AUDIT:,} Audit-Log records") ... print(f" Encryption : AES-256-GCM (NIST FIPS 197 / SP 800-38D)") ... print(' Search token scheme : HMAC-SHA256') ... print(" Approved keywords :",APPROVED_KEYWORDS) ... ... # ── Phase A: Data Owner ────────────────────────────────── ... ehr_df, audit_df = step1_generate_data() ... Load encrypt aes_key, hmac_secret, ehr_df, audit_df = step2_encrypt(ehr_df, audit_df) ... Based on the above, search_index, make_token are generated using the following step3_build_index function.Given the above, search_index, make_token are created by the following step3_build_index function. ... ... # ── Phase B: Investigator ──────────────────────────────── ... The step4_query function in MSearch returns the "matched_ids" keyword and search_index. ... ... if matched_ids: ... matched_ehr, matched_audit = step5_retrieve(matched_ids) ... step6_display(aes_key, matched_ehr, matched_audit) ... ... # ── Final File Manifest ────────────────────────────────── ... print(f"\n {'File':32s} {'Status':8s} {'Size':8s} Description") ... print(f" {'─'*32} {'─'*8} {'─'*8} {'─'*35}") ... manifest = [ ... ("ehr_records.csv", "Raw data plaintext EHR records (Step 1)"), ... Incidentally, the following line is incorrect: ("audit_logs.csv", "Raw plaintext Audit-Log records (Step 1)"), ... ("ehr_encrypted.json", "AES-256-GCM encrypted EHR (Step 2)"), ... ("audit_encrypted.json", AES-256-GCM encrypted Audit-Log (Step 2),), ... ("search_index.json", "HMAC-SHA256 token → Record ID index (Step 3)"), ... ("secret.key", "AES-256-GCM key — 256-bit (Step 2)"), ... ("hmac_secret.key", "HMAC-SHA256 signing secret (Step 2)"), ... ] ... For each of the lines in manifest: ... if os.path.exists(fname): ... size_kb = f"{os.path.getsize(fname)/1024:.0f} KB" ... status = "✓ OK" ... else: ... size_kb = "—" ... status = "✗ MISSING" ... print(f" {fname:32s} {status:8s} {size_kb:8s} {desc}") ... ... print('\n Total execution time :',time.time()-total_start,'.2f seconds)') ... print(" Encryption standard : AES-256-GCM ✓")