fix: address code quality issues across the project

- Add path traversal protection (safe_path utility) for RPA and APK extraction
- Fix rpa.py: rename read_util→read_until, add EOF check, handle bytes
  filenames, rename dir→output_dir parameter
- Fix stmts.py: mutable default args, variable shadowing, unused variable,
  return type annotation, error message, unpickler error handling
- Fix save.py: rstrip→removesuffix bug, remove redundant base64 imports
- Fix apk.py: tarfile path traversal protection
- Fix cli.py: verbose sets root logger, add sys import, nargs='+' for
  decompile, remove inline __import__
- Fix decompile.py: bare raise instead of raise e
- Fix translate.py: mutable default arg, add datetime import
- Fix safe_pickle.py: correct safe_loads docstring
- Add ecdsa optional dependency in pyproject.toml
- Export all public API from __init__.py

Author: cnfatal

pyproject.toml

@@ -19,6 +19,7 @@ dependencies = []
 
 [project.optional-dependencies]
 release = ["twine"]
+sign = ["ecdsa"]
 
 [project.urls]
 "Homepage" = "https://github.com/cnfatal/rpycdec"

src/rpycdec/__init__.py

@@ -1,4 +1,5 @@
 from .decompile import decompile, decompile_file
 from .cli import main
 from .rpa import extract_rpa
-from .save import extract_save, restore_save
+from .save import extract_save, restore_save, dump_save_info, generate_new_key
+from .translate import extract_translations

src/rpycdec/apk.py

@@ -6,12 +6,15 @@
 """
 
 import io
+import os
 import zipfile
 import tarfile
 import logging
 from pathlib import Path
 from typing import Optional
 
+from rpycdec.utils import safe_path
+
 logger = logging.getLogger(__name__)
 
 
@@ -172,7 +175,17 @@ def extract_apk(
                     )
 
                     with tarfile.open(fileobj=wrapped_stream, mode="r:gz") as tar:
-                        tar.extractall(path=private_output)
+                        # Path traversal protection
+                        for member in tar.getmembers():
+                            try:
+                                safe_path(str(private_output), member.name)
+                            except ValueError:
+                                logger.warning(
+                                    "Skipping path traversal attempt in tar: %s",
+                                    member.name,
+                                )
+                                continue
+                            tar.extract(member, path=private_output)
                         logger.info(
                             f"Extracted {len(tar.getmembers())} files from private.mp3"
                         )

src/rpycdec/cli.py

@@ -1,6 +1,8 @@
 import argparse
 import logging
 import os
+import sys
+
 from rpycdec.decompile import decompile
 from rpycdec.rpa import extract_rpa
 from rpycdec.save import extract_save, restore_save, dump_save_info, generate_new_key
@@ -30,7 +32,7 @@ def extract_rpa_files(srcs: list[str], **kwargs):
     for src in srcs:
         with open(src, "rb") as f:
             output_path = kwargs.get("output") or os.path.dirname(src)
-            extract_rpa(f, dir=output_path)
+            extract_rpa(f, output_dir=output_path)
 
 
 def run_extract_translations(
@@ -84,7 +86,7 @@ def main():
     )
 
     decompile_parser = subparsers.add_parser("decompile", help="decompile rpyc file")
-    decompile_parser.add_argument("src", nargs=1, help="rpyc file or directory")
+    decompile_parser.add_argument("src", nargs="+", help="rpyc file or directory")
     decompile_parser.add_argument(
         "--output",
         "-o",
@@ -222,7 +224,7 @@ def main():
 
     args = argparser.parse_args()
     if args.verbose:
-        logger.setLevel(logging.DEBUG)
+        logging.getLogger().setLevel(logging.DEBUG)
     if not args.command:
         argparser.print_help()
         return
@@ -231,6 +233,6 @@ def main():
     if args.command in ("decompile", "save", "unrpa") and not os.environ.get(
         "RPYCDEC_NO_WARNING"
     ):
-        print(SECURITY_WARNING, file=__import__("sys").stderr)
+        print(SECURITY_WARNING, file=sys.stderr)
 
     args.func(args)

src/rpycdec/decompile.py

@@ -26,7 +26,7 @@ def decompile_file(input_file, output_path=None, **kwargs):
         code = util.get_code(stmt)
     except Exception as e:
         logger.error("decode file %s failed: %s", input_file, e)
-        raise e
+        raise
     utils.write_file(output_path, code)
     logger.info("decompile %s -> %s", input_file, output_path)
 

src/rpycdec/rpa.py

@@ -1,15 +1,25 @@
+import logging
 import os
 import zlib
 from io import BufferedIOBase
 
 from rpycdec.safe_pickle import rpa_loads
+from rpycdec.utils import safe_path
 
+logger = logging.getLogger(__name__)
 
-def read_util(data: BufferedIOBase, util: int = 0x00) -> bytes:
+
+def read_until(data: BufferedIOBase, delimiter: int = 0x00) -> bytes:
+    """Read bytes from stream until delimiter is found.
+
+    Raises ValueError on unexpected EOF.
+    """
     content = bytearray()
     while True:
         c = data.read(1)
-        if c[0] == util:
+        if not c:
+            raise ValueError("Unexpected EOF while reading stream")
+        if c[0] == delimiter:
             break
         content += c
     return content
@@ -23,14 +33,13 @@ def start_to_bytes(left: list | None) -> bytes:
     return left[0].encode("latin-1")
 
 
-def extract_rpa(r: BufferedIOBase, dir: str | None = None):
-    dir = dir or "."
-    magic = read_util(r, 0x20)
+def extract_rpa(r: BufferedIOBase, output_dir: str | None = None):
+    output_dir = output_dir or "."
+    magic = read_until(r, 0x20)
     if magic != b"RPA-3.0":
-        print("Not a Ren'Py archive.")
-        return
-    index_offset = int(read_util(r, 0x20), 16)
-    key = int(read_util(r, 0x0A).decode(), 16)
+        raise ValueError("Not a Ren'Py RPA-3.0 archive.")
+    index_offset = int(read_until(r, 0x20), 16)
+    key = int(read_until(r, 0x0A).decode(), 16)
 
     # read index
     r.seek(index_offset)
@@ -43,6 +52,10 @@ def extract_rpa(r: BufferedIOBase, dir: str | None = None):
         ]
 
     for filename, entries in index.items():
+        # Handle bytes filenames from Python 2 era archives
+        if isinstance(filename, bytes):
+            filename = filename.decode("utf-8", errors="surrogateescape")
+
         data = bytearray()
         for offset, dlen, start in entries:
             r.seek(offset)
@@ -51,11 +64,19 @@ def extract_rpa(r: BufferedIOBase, dir: str | None = None):
                 if block.startswith(start):
                     block = block[len(start) :]
                 else:
-                    print("Warning: %s does not start with %s" % (filename, start))
+                    logger.warning(
+                        "%s does not start with expected prefix %s", filename, start
+                    )
             data += block
 
-        filename = os.path.join(dir, filename)
-        os.makedirs(os.path.dirname(filename), exist_ok=True)
-        with open(filename, "wb") as f:
-            print("extracting: ", filename)
+        # Path traversal protection
+        try:
+            dest = safe_path(output_dir, filename)
+        except ValueError:
+            logger.warning("Skipping path traversal attempt: %s", filename)
+            continue
+
+        os.makedirs(os.path.dirname(dest), exist_ok=True)
+        with open(dest, "wb") as f:
+            logger.info("extracting: %s", dest)
             f.write(data)

src/rpycdec/safe_pickle.py

@@ -219,7 +219,10 @@ def find_class(self, module: str, name: str) -> Any:
 
 
 def safe_loads(data: bytes, **kwargs: Any) -> Any:
-    """Safely unpickle data using SafeUnpickler (strict mode — raises on unknown classes)."""
+    """Safely unpickle data using SafeUnpickler.
+
+    Unknown classes are substituted with DummyClass placeholders.
+    """
     return SafeUnpickler(io.BytesIO(data), **kwargs).load()
 
 

src/rpycdec/save.py

@@ -498,7 +498,7 @@ def restore_save(
         key_file: Path to security_keys.txt for re-signing (optional)
     """
     if not output_file:
-        output_file = extracted_dir.rstrip("/").rstrip(".extracted") + ".restored.save"
+        output_file = extracted_dir.rstrip("/").removesuffix(".extracted") + ".restored.save"
 
     print(f"Restoring save file from: {extracted_dir}")
     print(f"Output file: {output_file}")
@@ -604,8 +604,6 @@ def sign_data(log_data: bytes, key_file: str) -> str:
 
     def encode_line(kind: str, key: bytes, sig: bytes = b"") -> str:
         """Encode a signature line in Ren'Py format."""
-        import base64
-
         key_b64 = base64.b64encode(key).decode("ascii")
         if sig:
             sig_b64 = base64.b64encode(sig).decode("ascii")
@@ -614,8 +612,6 @@ def encode_line(kind: str, key: bytes, sig: bytes = b"") -> str:
 
     def decode_line(line: str) -> Tuple[str, bytes, bytes]:
         """Decode a signature line from Ren'Py format."""
-        import base64
-
         parts = line.strip().split()
         if len(parts) < 2:
             return "", b"", b""
@@ -661,8 +657,6 @@ def generate_new_key(key_file: str) -> str:
     except ImportError:
         raise ImportError("ecdsa library required. Install with: pip install ecdsa")
 
-    import base64
-
     sk = ecdsa.SigningKey.generate(curve=ecdsa.NIST256p)
     vk = sk.verifying_key
 

src/rpycdec/stmts.py

@@ -41,38 +41,43 @@ def read_rpyc_data(file: io.BufferedReader, slot):
     return zlib.decompress(data)
 
 
-def load(data: io.BufferedReader, slots: list[int] = [1, 2], **kwargs) -> Node | None:
+def load(data: io.BufferedReader, slots: list[int] | None = None, **kwargs) -> list[Node] | None:
+    """Load Ren'Py AST from a .rpyc file.
+
+    Tries each slot in order. Slot 1 is the original script,
+    slot 2 is the pre-translated version.
+    """
+    if slots is None:
+        slots = [1, 2]
     # 1 is statements before translation, 2 is after translation.
     for slot in slots:
         try:
             bindata = read_rpyc_data(data, slot)
-        except Exception as e:
-            logger.warning(f"Failed to read slot {slot}: {e}")
-            data.seek(0)
-            continue
-        if bindata:
+            if not bindata:
+                continue
+
             if kwargs.get("dis", False):
                 logger.info("Disassembling rpyc file...")
                 pickletools.dis(bindata)
 
             unpickler = SafeUnpickler(
                 io.BytesIO(bindata), encoding="utf-8", errors="surrogateescape"
             )
-            data, stmts = unpickler.load()
-
-            key = data.get("key", "unlocked")  # type: ignore
+            metadata, stmts = unpickler.load()
             return stmts
-    raise Exception("Unsupported file format or invalid file")
+        except Exception as e:
+            logger.warning(f"Failed to read slot {slot}: {e}")
+            data.seek(0)
+            continue
+    raise ValueError("Unsupported file format or invalid file")
 
 
-def load_file(filename, **kwargs) -> Node | None:
-    """
-    load renpy code from rpyc file and return ast tree.
-    """
+def load_file(filename, **kwargs) -> list[Node] | None:
+    """Load Ren'Py AST from a .rpyc/.rpymc file."""
     ext = path.splitext(filename)[1]
     if ext in [".rpy", ".rpym"]:
         raise NotImplementedError(
-            "unsupport for pase rpy file or use renpy.parser.parse() in renpy's SDK"
+            "Parsing .rpy files is not supported. Use renpy.parser.parse() from Ren'Py's SDK."
         )
     # slot 2 is for pre-translated scripts, slot 1 is for normal scripts
     slots = [2, 1] if kwargs.get("pre_translated", False) else [1, 2]

src/rpycdec/translate.py

@@ -1,4 +1,5 @@
 import collections
+import datetime
 import hashlib
 import logging
 import os
@@ -201,8 +202,10 @@ def _generate_identifier(self, node: renpy.ast.Node) -> str:
 # ============================================================================
 
 
-def unique_identifier(label: str | None, digest: str, existing_identifiers: set = set()) -> str:
+def unique_identifier(label: str | None, digest: str, existing_identifiers: set | None = None) -> str:
     """Generate a unique translation identifier"""
+    if existing_identifiers is None:
+        existing_identifiers = set()
     if label is None:
         base = digest
     else:
@@ -291,7 +294,7 @@ def write_dialogue_translations(
         os.makedirs(os.path.dirname(tl_path) if os.path.dirname(tl_path) else output_dir, exist_ok=True)
 
         with open(tl_path, "w", encoding="utf-8") as f:
-            f.write(f"# TODO: Translation updated at {__import__('datetime').datetime.now().isoformat()}\n\n")
+            f.write(f"# TODO: Translation updated at {datetime.datetime.now().isoformat()}\n\n")
 
             for item in items:
                 # Write source file location comment
@@ -346,7 +349,7 @@ def write_string_translations(
     tl_path = os.path.join(output_dir, "strings.rpy")
 
     with open(tl_path, "w", encoding="utf-8") as f:
-        f.write(f"# TODO: Translation updated at {__import__('datetime').datetime.now().isoformat()}\n\n")
+        f.write(f"# TODO: Translation updated at {datetime.datetime.now().isoformat()}\n\n")
         f.write(f"translate {language} strings:\n\n")
 
         for item in strings: