Bulk Replace Text Hidden in Avada [fusion_code] Base64 Blocks (WordPress + Python)

The key limitation is not "safety", it is visibility. When Avada stores content as Base64 inside Base64 blocks in wp_posts.post_content, decode them, apply string replacements, re-encode them, and optionally update rows. Why: Many WordPress search/replace tools cannot replace text hidden inside Base64 payloads because the text is not stored as plain strings in the database. Usage: # Dry run (report only) python3 fuse_b64_update.py # Apply changes (perform DB updates) python3 fuse_b64_update.py --apply Credentials: Use environment variables instead of hardcoding: WP_DB_HOST, WP_DB_USER, WP_DB_PASS, WP_DB_NAME Optional overrides: WP_DB_TABLE, WP_DB_COLUMN, WP_DB_ID_COLUMN """ from __future__ import annotations import argparse import base64 import os import re import sys from dataclasses import dataclass from typing import List, Tuple import pymysql # ========================= # Configuration (safe) # ========================= DB_HOST = os.getenv("WP_DB_HOST", "localhost") DB_USER = os.getenv("WP_DB_USER", "") DB_PASS = os.getenv("WP_DB_PASS", "") DB_NAME = os.getenv("WP_DB_NAME", "") TABLE = os.getenv("WP_DB_TABLE", "wp_posts") COLUMN = os.getenv("WP_DB_COLUMN", "post_content") ID_COLUMN = os.getenv("WP_DB_ID_COLUMN", "ID") # EDIT THIS ONLY: # Mapping of "find" -> "replace" REPLACEMENTS = { # Domain name changes (most common): # "oldcompany.com": "newcompany.ca", # Phone number updates: # "(604) 555-1234": "778-555-5678", # Email address changes: # "[email protected]": "[email protected]", # Company name rebranding: # "ABC Services Ltd.": "XYZ Solutions Inc.", # Full URL updates: # "http://oldsite.com": "https://newsite.ca", } # Regex to capture Base64 inside FUSION_CODE_RE = re.compile( r"(\\)\\s*([A-Za-z0-9+/=\\r\\n]+?)\\s*(\\)", re.DOTALL | re.IGNORECASE, ) # Limit number of sample previews per row (avoid huge output) SAMPLES_PER_ROW = 3 SNIPPET_LEN = 260 @dataclass class Preview: before: str after: str def decode_b64(b64_text: str) -> Tuple[str, bool]: """Decode Base64 safely. Returns (decoded_text, success).""" try: clean = "".join(b64_text.split()) decoded_bytes = base64.b64decode(clean, validate=False) decoded = decoded_bytes.decode("utf-8", errors="strict") return decoded, True except Exception: return "", False def encode_b64(text: str) -> str: """Encode text to Base64 without newlines.""" return base64.b64encode(text.encode("utf-8")).decode("ascii") def apply_replacements(decoded: str) -> Tuple[str, bool]: """Apply REPLACEMENTS mapping. Returns (new_text, changed).""" new_text = decoded for old, new_val in REPLACEMENTS.items(): new_text = new_text.replace(old, new_val) return new_text, (new_text != decoded) def process_content(content: str) -> Tuple[str, int, List[Preview]]: """ Process post content: - find Base64 blocks - decode - apply replacements - re-encode Returns: (new_content, number_of_changed_blocks, previews) """ changes = 0 previews: List[Preview] = [] def replacer(match: re.Match) -> str: nonlocal changes, previews prefix, b64payload, suffix = match.group(1), match.group(2), match.group(3) decoded, ok = decode_b64(b64payload) if not ok: return match.group(0) # leave unchanged new_decoded, changed = apply_replacements(decoded) if not changed: return match.group(0) changes += 1 if len(previews) < SAMPLES_PER_ROW: before_snip = decoded[:SNIPPET_LEN].replace("\\n", "\\\\n") after_snip = new_decoded[:SNIPPET_LEN].replace("\\n", "\\\\n") previews.append(Preview(before=before_snip, after=after_snip)) new_b64 = encode_b64(new_decoded) return f"{prefix}{new_b64}{suffix}" new_content = FUSION_CODE_RE.sub(replacer, content or "") return new_content, changes, previews def require_db_settings() -> None: missing = [] if not DB_USER: missing.append("WP_DB_USER") if not DB_NAME: missing.append("WP_DB_NAME") # password can be blank in some local setups, so we do not force it if missing: print("ERROR: Missing DB settings:", ", ".join(missing)) print("Set env vars: WP_DB_HOST, WP_DB_USER, WP_DB_PASS, WP_DB_NAME") sys.exit(1) def connect_db(): try: return pymysql.connect( host=DB_HOST, user=DB_USER, password=DB_PASS, database=DB_NAME, charset="utf8mb4", ) except Exception as e: print("ERROR: could not connect to database:", e) sys.exit(1) def main() -> None: parser = argparse.ArgumentParser( description="Decode Avada Base64 blocks and apply replacements." ) parser.add_argument( "--apply", action="store_true", help="Apply updates to the database (default: dry-run only).", ) args = parser.parse_args() require_db_settings() mode = "APPLY MODE" if args.apply else "DRY RUN MODE" print(f"{mode} | Replacements: {REPLACEMENTS}") conn = connect_db() total_rows = 0 total_rows_changed = 0 total_blocks_changed = 0 try: with conn.cursor() as cur: cur.execute( f"SELECT {ID_COLUMN}, {COLUMN} FROM {TABLE} WHERE {COLUMN} LIKE %s", ("%%",), ) rows = cur.fetchall() total_rows = len(rows) print(f"Found {total_rows} rows containing ...") for row_id, content in rows: new_content, blocks_changed, previews = process_content(content or "") if blocks_changed <= 0: continue total_rows_changed += 1 total_blocks_changed += blocks_changed print("-" * 80) print(f"Row ID: {row_id} | Changed fusion_code blocks: {blocks_changed}") for i, pv in enumerate(previews, start=1): print(f" Preview #{i} (truncated):") print(" BEFORE:", pv.before) print(" AFTER: ", pv.after) print() if args.apply: cur.execute( f"UPDATE {TABLE} SET {COLUMN} = %s WHERE {ID_COLUMN} = %s", (new_content, row_id), ) print(f" -> Row {row_id} UPDATED.") else: print(f" -> DRY RUN: no DB update performed for row {row_id}.") if args.apply: conn.commit() finally: conn.close() print("=" * 80) print(f"Rows scanned: {total_rows}") print(f"Rows with changes: {total_rows_changed}") print(f"Total fusion_code blocks changed: {total_blocks_changed}") print("Done.") if __name__ == "__main__": main()