diff --git a/.github/workflows/container-local-test.yml b/.github/workflows/container-local-test.yml index 053f794b..a1423f6c 100644 --- a/.github/workflows/container-local-test.yml +++ b/.github/workflows/container-local-test.yml @@ -21,11 +21,11 @@ jobs: steps: - name: Checkout Repository - uses: actions/checkout@v4 + uses: actions/checkout@v6 # Setup and build the python package - name: Set up Python - uses: actions/setup-python@v5 + uses: actions/setup-python@v6 with: python-version: "3.9.x" @@ -38,12 +38,12 @@ jobs: run: make dist - name: Setup Docker buildx - uses: docker/setup-buildx-action@v3 + uses: docker/setup-buildx-action@v4 # Build Docker image with Buildx - Base - name: Build Docker Image - No Entrypoint id: build-no-ep - uses: docker/build-push-action@v5 + uses: docker/build-push-action@v7 with: context: . push: false @@ -54,7 +54,7 @@ jobs: # Build Docker image with Buildx - Jenkins - name: Build Docker Image - Jenkins id: build-je - uses: docker/build-push-action@v5 + uses: docker/build-push-action@v7 with: context: . push: false @@ -65,7 +65,7 @@ jobs: # Build Docker image with Buildx - Entrypoint - name: Build Docker Image - With Entrypoint id: build-with-ep - uses: docker/build-push-action@v5 + uses: docker/build-push-action@v7 with: context: . push: false @@ -91,6 +91,7 @@ jobs: docker image ls -a docker run ${{ env.IMAGE_NAME }} version docker run ${{ env.IMAGE_NAME }} utils fast + echo "Running file scan on tests" docker run -e SCANOSS_API_KEY="${{ secrets.SC_API_KEY }}" -v "$(pwd)":"/scanoss" ${{ env.IMAGE_NAME }} scan -o results.json tests id_count=$(cat results.json | grep '"id":' | wc -l) echo "ID Count: $id_count" @@ -98,10 +99,20 @@ jobs: echo "Error: Scan test did not produce any results. Failing" exit 1 fi + echo "Running fingerprinting on tests" docker run -v "$(pwd)":"/scanoss" ${{ env.IMAGE_NAME }} wfp --skip-headers -o fingers.wfp tests wfp_count=$(cat fingers.wfp | grep 'file=' | wc -l) echo "WFP Count: $wfp_count" if [[ $wfp_count -lt 1 ]]; then echo "Error: WFP test did not produce any results. Failing" exit 1 - fi \ No newline at end of file + fi + echo "Running dependency scan on tests" + docker run -e SCANOSS_API_KEY="${{ secrets.SC_API_KEY }}" -v "$(pwd)":"/scanoss" ${{ env.IMAGE_NAME }} scan --dependencies-only -o results.json tests + dep_count=$(python3 -c "import json,sys; d=json.load(open('results.json')); print(sum(len(v.get('dependencies',[])) for f in d.values() for v in f))" 2>/dev/null || echo 0) + echo "Dependency count: $dep_count" + if [[ $dep_count -lt 1 ]]; then + echo "Error: Dependency scan did not find any dependencies. Failing" + exit 1 + fi + diff --git a/.github/workflows/container-publish-ghcr.yml b/.github/workflows/container-publish-ghcr.yml index 84aebab0..7874c90a 100644 --- a/.github/workflows/container-publish-ghcr.yml +++ b/.github/workflows/container-publish-ghcr.yml @@ -24,11 +24,11 @@ jobs: steps: - name: Checkout Repository - uses: actions/checkout@v4 + uses: actions/checkout@v6 # Setup and build python package - name: Set up Python - uses: actions/setup-python@v5 + uses: actions/setup-python@v6 with: python-version: '3.9.x' @@ -42,12 +42,12 @@ jobs: # Add support for more platforms with QEMU - name: Set up QEMU - uses: docker/setup-qemu-action@v3 + uses: docker/setup-qemu-action@v4 # Workaround: https://github.com/docker/build-push-action/issues/461 # uses: docker/setup-buildx-action@79abd3f86f79a9d68a23c75a09a9a85889262adf - name: Setup Docker buildx - uses: docker/setup-buildx-action@v3 + uses: docker/setup-buildx-action@v4 # Login against a Docker registry except on PR - name: Log into registry ${{ env.REGISTRY }} @@ -60,14 +60,14 @@ jobs: # Extract metadata (tags, labels) for Docker - name: Extract Docker metadata - no entrypoint id: meta-ne - uses: docker/metadata-action@v4 + uses: docker/metadata-action@v6 with: images: "${{ env.REGISTRY }}/${{ env.IMAGE_NAME_BASE }}" # Build and push Docker image with Buildx (don't push on PR) - name: Build and push Docker image - Base (no entrypoint) id: build-and-push-ne - uses: docker/build-push-action@v5 + uses: docker/build-push-action@v7 with: context: . push: ${{ github.event_name != 'pull_request' }} @@ -82,14 +82,14 @@ jobs: # Extract metadata (tags, labels) for Docker - name: Extract Docker metadata - jenkins id: meta-je - uses: docker/metadata-action@v4 + uses: docker/metadata-action@v6 with: images: "${{ env.REGISTRY }}/${{ env.IMAGE_JENKINS }}" # Build and push Docker image with Buildx (don't push on PR) - name: Build and push Docker image - Jenkins id: build-and-push-je - uses: docker/build-push-action@v5 + uses: docker/build-push-action@v7 with: context: . push: ${{ github.event_name != 'pull_request' }} @@ -104,14 +104,14 @@ jobs: # Extract metadata (tags, labels) for Docker - name: Extract Docker metadata - entrypoint id: meta - uses: docker/metadata-action@v4 + uses: docker/metadata-action@v6 with: images: "${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}" # Build and push Docker image with Buildx (don't push on PR) - name: Build and push Docker image - EP (entrypoint) id: build-and-push - uses: docker/build-push-action@v5 + uses: docker/build-push-action@v7 with: context: . push: ${{ github.event_name != 'pull_request' }} diff --git a/.github/workflows/lint.yml b/.github/workflows/lint.yml index 11367ddc..468a6b97 100644 --- a/.github/workflows/lint.yml +++ b/.github/workflows/lint.yml @@ -10,12 +10,12 @@ jobs: steps: - name: Checkout Code - uses: actions/checkout@v3 + uses: actions/checkout@v6 with: fetch-depth: 0 - name: Set up Python - uses: actions/setup-python@v4 + uses: actions/setup-python@v6 with: python-version: "3.9" diff --git a/.github/workflows/python-local-test.yml b/.github/workflows/python-local-test.yml index 59aa6813..1e30e67f 100644 --- a/.github/workflows/python-local-test.yml +++ b/.github/workflows/python-local-test.yml @@ -20,10 +20,10 @@ jobs: build: runs-on: ubuntu-latest steps: - - uses: actions/checkout@v4 + - uses: actions/checkout@v6 - name: Set up Python - uses: actions/setup-python@v5 + uses: actions/setup-python@v6 with: python-version: "3.9.x" @@ -36,7 +36,7 @@ jobs: run: make dist - name: Install Test Package - uses: nick-fields/retry@v3 + uses: nick-fields/retry@v4 with: timeout_minutes: 2 retry_wait_seconds: 10 @@ -103,6 +103,18 @@ jobs: exit 1 fi + - name: Run Tests HPFM + run: | + which scanoss-py + scanoss-py version + scanoss-py fh -o hpfm-hashes.json tests + hash_names=$(cat hpfm-hashes.json | grep 'sim_hash_names') + echo "Sim Hash: $hash_names" + if [ "$hash_names" = "" ]; then + echo "Error: Folder hash failed to generate HPFM hashes. Failing." + exit 1 + fi + - name: Run Unit Tests run: | python -m unittest diff --git a/.github/workflows/python-publish-pypi.yml b/.github/workflows/python-publish-pypi.yml index 1dc8b571..04c5f592 100644 --- a/.github/workflows/python-publish-pypi.yml +++ b/.github/workflows/python-publish-pypi.yml @@ -14,10 +14,10 @@ jobs: deploy: runs-on: ubuntu-latest steps: - - uses: actions/checkout@v4 + - uses: actions/checkout@v6 - name: Set up Python - uses: actions/setup-python@v5 + uses: actions/setup-python@v6 with: python-version: '3.9.x' @@ -69,7 +69,7 @@ jobs: - name: Create Draft Release ${{ github.ref_type }} - ${{ github.ref_name }} if: github.ref_type == 'tag' && startsWith(github.ref_name, 'v') - uses: softprops/action-gh-release@v1 + uses: softprops/action-gh-release@v3 with: draft: true files: dist/* @@ -79,15 +79,15 @@ jobs: needs: [deploy] runs-on: ubuntu-latest steps: - - uses: actions/checkout@v4 + - uses: actions/checkout@v6 - name: Set up Python - uses: actions/setup-python@v5 + uses: actions/setup-python@v6 with: python-version: '3.9.x' - name: Install Remote Package - uses: nick-fields/retry@v3 + uses: nick-fields/retry@v4 with: timeout_minutes: 3 retry_wait_seconds: 10 diff --git a/.github/workflows/python-publish-testpypi.yml b/.github/workflows/python-publish-testpypi.yml index d44c02a0..41121e0e 100644 --- a/.github/workflows/python-publish-testpypi.yml +++ b/.github/workflows/python-publish-testpypi.yml @@ -13,10 +13,10 @@ jobs: deploy: runs-on: ubuntu-latest steps: - - uses: actions/checkout@v4 + - uses: actions/checkout@v6 - name: Set up Python - uses: actions/setup-python@v5 + uses: actions/setup-python@v6 with: python-version: '3.9.x' @@ -63,10 +63,10 @@ jobs: runs-on: ubuntu-latest steps: - - uses: actions/checkout@v4 + - uses: actions/checkout@v6 - name: Set up Python - uses: actions/setup-python@v5 + uses: actions/setup-python@v6 with: python-version: '3.9.x' diff --git a/.github/workflows/scanoss.yml b/.github/workflows/scanoss.yml index d9dcb338..605032ae 100644 --- a/.github/workflows/scanoss.yml +++ b/.github/workflows/scanoss.yml @@ -20,11 +20,11 @@ jobs: steps: - name: Checkout code - uses: actions/checkout@v4 + uses: actions/checkout@v6 - name: Run SCANOSS Code Scan id: scanoss-code-scan-step - uses: scanoss/code-scan-action@v1 + uses: scanoss/gha-code-scan@v1 with: policies: undeclared api.url: https://api.scanoss.com/scan/direct diff --git a/.github/workflows/version-tag.yml b/.github/workflows/version-tag.yml index 55afecb0..8c45614a 100644 --- a/.github/workflows/version-tag.yml +++ b/.github/workflows/version-tag.yml @@ -16,12 +16,13 @@ jobs: version-tagging: runs-on: ubuntu-latest steps: - - uses: actions/checkout@v4 + - uses: actions/checkout@v6 with: fetch-depth: "0" token: ${{ secrets.SC_GH_TAG_TOKEN }} + persist-credentials: false - name: Set up Python - uses: actions/setup-python@v5 + uses: actions/setup-python@v6 with: python-version: '3.9.x' - name: Determine Tag @@ -34,9 +35,12 @@ jobs: - name: Apply Tag if: ${{ inputs.run_for_real }} id: taggerApply + env: + TAG_TOKEN: ${{ secrets.SC_GH_TAG_TOKEN }} run: | echo "Applying tag ${{env.package_app_version}} ..." git tag "${{env.package_app_version}}" echo "Pushing changes..." + git remote set-url origin "https://x-access-token:${TAG_TOKEN}@github.com/${{ github.repository }}" git push --tags diff --git a/CHANGELOG.md b/CHANGELOG.md index 6b85e6c8..edf2e3a0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,19 @@ All notable changes to this project will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [Unreleased] + +## [1.54.0] - 2026-07-07 +### Added +- Added `--scan-root` to specify the scan root to load scanoss.json from (Commands: `scan`, `wfp`, `dependencies`, `folder-scan`, `folder-hash`) +- Added a priority list for loading scanoss.json (1. --settings 2. Scan directory (`--scan-root` if specified) 3. CWD) + +## [1.53.2] - 2026-07-07 +### Fixed +- No longer reports every HTTP 503 as "service limits being exceeded", which misled users into thinking they had hit a rate limit during unrelated service outages. Rate limits (HTTP 429) are now distinguished from generic service unavailability (HTTP 503), and the actual server response body is surfaced in the error message. +### Changed +- Honour the `Retry-After` header on HTTP 429/503 responses, retrying with backoff (capped at 60s) instead of aborting immediately on the first 503 + ## [1.53.1] - 2026-06-04 ### Added - Added `--apiurl` option to the `folder-scan` command to configure a custom SCANOSS API base URL (consistent with the `scan` command) @@ -894,3 +907,5 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 [1.52.1]: https://github.com/scanoss/scanoss.py/compare/v1.52.0...v1.52.1 [1.53.0]: https://github.com/scanoss/scanoss.py/compare/v1.52.1...v1.53.0 [1.53.1]: https://github.com/scanoss/scanoss.py/compare/v1.53.0...v1.53.1 +[1.53.2]: https://github.com/scanoss/scanoss.py/compare/v1.53.1...v1.53.2 +[1.54.0]: https://github.com/scanoss/scanoss.py/compare/v1.53.2...v1.54.0 diff --git a/src/scanoss/__init__.py b/src/scanoss/__init__.py index ac0ecd8e..de41286e 100644 --- a/src/scanoss/__init__.py +++ b/src/scanoss/__init__.py @@ -22,4 +22,4 @@ THE SOFTWARE. """ -__version__ = '1.53.1' +__version__ = '1.54.0' diff --git a/src/scanoss/cli.py b/src/scanoss/cli.py index 9c1ddffc..6695e01d 100644 --- a/src/scanoss/cli.py +++ b/src/scanoss/cli.py @@ -127,6 +127,11 @@ def setup_args() -> None: # noqa: PLR0912, PLR0915 ) p_scan.set_defaults(func=scan) p_scan.add_argument('scan_dir', metavar='FILE/DIR', type=str, nargs='?', help='A file or folder to scan') + p_scan.add_argument( + '--scan-root', '-scr', type=str, + help='Scan root directory. FILE/DIR is treated as relative to this root, ' + 'and scanoss.json is loaded from it.', + ) p_scan.add_argument('--wfp', '-w', type=str, help='Scan a WFP File instead of a folder (optional)') p_scan.add_argument('--dep', '-p', type=str, help='Use a dependency file instead of a folder (optional)') p_scan.add_argument( @@ -239,6 +244,11 @@ def setup_args() -> None: # noqa: PLR0912, PLR0915 ) p_wfp.set_defaults(func=wfp) p_wfp.add_argument('scan_dir', metavar='FILE/DIR', type=str, nargs='?', help='A file or folder to scan') + p_wfp.add_argument( + '--scan-root', '-scr', type=str, + help='Scan root directory. scan_dir is treated as relative to this root, ' + 'and scanoss.json is loaded from it.', + ) p_wfp.add_argument( '--stdin', '-s', @@ -254,7 +264,12 @@ def setup_args() -> None: # noqa: PLR0912, PLR0915 description=f'Produce dependency file summary: {__version__}', help='Scan source code for dependencies, but do not decorate them', ) - p_dep.add_argument('scan_loc', metavar='FILE/DIR', type=str, nargs='?', help='A file or folder to scan') + p_dep.add_argument('scan_dir', metavar='FILE/DIR', type=str, nargs='?', help='A file or folder to scan') + p_dep.add_argument( + '--scan-root', '-scr', type=str, + help='Scan root directory. FILE/DIR is treated as relative to this root, ' + 'and scanoss.json is loaded from it.', + ) p_dep.add_argument( '--container', type=str, @@ -1033,6 +1048,11 @@ def setup_args() -> None: # noqa: PLR0912, PLR0915 f'(optional - default: {DEFAULT_HFH_MIN_ACCEPTED_SCORE})' ), ) + p_folder_scan.add_argument( + '--scan-root', '-scr', type=str, + help='Scan root directory. FILE/DIR is treated as relative to this root, ' + 'and scanoss.json is loaded from it.', + ) p_folder_scan.set_defaults(func=folder_hashing_scan) # Sub-command: folder-hash @@ -1057,6 +1077,11 @@ def setup_args() -> None: # noqa: PLR0912, PLR0915 default=DEFAULT_HFH_DEPTH, help=f'Defines how deep to hash the root directory (optional - default {DEFAULT_HFH_DEPTH})', ) + p_folder_hash.add_argument( + '--scan-root', '-scr', type=str, + help='Scan root directory. FILE/DIR is treated as relative to this root, ' + 'and scanoss.json is loaded from it.', + ) p_folder_hash.set_defaults(func=folder_hash) # Sub-command: delta @@ -1457,21 +1482,13 @@ def wfp(parser, args): print_stderr('Please specify a file/folder or STDIN (--stdin)') parser.parse_args([args.subparser, '-h']) sys.exit(1) + validate_scan_root(args) if args.strip_hpsm and not args.hpsm and not args.quiet: print_stderr('Warning: --strip-hpsm option supplied without enabling HPSM (--hpsm). Ignoring.') if args.output: initialise_empty_file(args.output) - # Load scan settings - scanoss_settings = None - if not args.skip_settings_file: - scanoss_settings = ScanossSettings(debug=args.debug, trace=args.trace, quiet=args.quiet) - try: - scanoss_settings.load_json_file(args.settings, args.scan_dir) - except ScanossSettingsError as e: - print_stderr(f'Error: {e}') - sys.exit(1) - + scanoss_settings = get_scanoss_settings_from_args(args) scan_options = 0 if args.skip_snippets else ScanType.SCAN_SNIPPETS.value # Skip snippet generation or not scanner = Scanner( debug=args.debug, @@ -1497,15 +1514,25 @@ def wfp(parser, args): contents = sys.stdin.buffer.read() scanner.wfp_contents(args.stdin, contents, args.output) elif args.scan_dir: - if not os.path.exists(args.scan_dir): - print_stderr(f'Error: File or folder specified does not exist: {args.scan_dir}.') + scan_root = args.scan_root + effective_path = os.path.join(scan_root, args.scan_dir) if scan_root else args.scan_dir + relative_target = None + if scan_root: + relative_target = os.path.relpath( + Path(effective_path).resolve(), + Path(scan_root).resolve(), + ) + if relative_target == '.': + relative_target = None + if not os.path.exists(effective_path): + print_stderr(f'Error: File or folder specified does not exist: {effective_path}.') sys.exit(1) - if os.path.isdir(args.scan_dir): - scanner.wfp_folder(args.scan_dir, args.output) - elif os.path.isfile(args.scan_dir): - scanner.wfp_file(args.scan_dir, args.output) + if os.path.isdir(effective_path): + scanner.wfp_folder(scan_root or args.scan_dir, args.output, filter_path=relative_target) + elif os.path.isfile(effective_path): + scanner.wfp_file(effective_path, args.output, file_id=relative_target) else: - print_stderr(f'Error: Path specified is neither a file or a folder: {args.scan_dir}.') + print_stderr(f'Error: Path specified is neither a file or a folder: {effective_path}.') sys.exit(1) else: print_stderr('No action found to process') @@ -1575,28 +1602,9 @@ def scan(parser, args): # noqa: PLR0912, PLR0915 if args.identify and args.settings: print_stderr('ERROR: Cannot specify both --identify and --settings options.') sys.exit(1) - if args.settings and args.skip_settings_file: - print_stderr('ERROR: Cannot specify both --settings and --skip-file-settings options.') - sys.exit(1) + validate_scan_root(args) # Figure out which settings (if any) to load before processing - scanoss_settings = None - if not args.skip_settings_file: - scanoss_settings = ScanossSettings(debug=args.debug, trace=args.trace, quiet=args.quiet) - try: - if args.identify: - scanoss_settings.load_json_file(args.identify, args.scan_dir).set_file_type('legacy').set_scan_type( - 'identify' - ) - elif args.ignore: - scanoss_settings.load_json_file(args.ignore, args.scan_dir).set_file_type('legacy').set_scan_type( - 'blacklist' - ) - else: - scanoss_settings.load_json_file(args.settings, args.scan_dir).set_file_type('new') - - except ScanossSettingsError as e: - print_stderr(f'Error: {e}') - sys.exit(1) + scanoss_settings = get_scanoss_settings_from_args(args) if args.dep: if not os.path.exists(args.dep) or not os.path.isfile(args.dep): print_stderr(f'Specified --dep file does not exist or is not a file: {args.dep}') @@ -1717,31 +1725,43 @@ def scan(parser, args): # noqa: PLR0912, PLR0915 if not scanner.scan_files_with_options(file_list, args.dep, scanner.winnowing.file_map): sys.exit(1) elif args.scan_dir: - if not os.path.exists(args.scan_dir): - print_stderr(f'Error: File or folder specified does not exist: {args.scan_dir}.') + scan_root = args.scan_root + effective_path = os.path.join(scan_root, args.scan_dir) if scan_root else args.scan_dir + relative_target = None + if scan_root: + relative_target = os.path.relpath( + Path(effective_path).resolve(), + Path(scan_root).resolve(), + ) + if relative_target == '.': + relative_target = None + if not os.path.exists(effective_path): + print_stderr(f'Error: File or folder specified does not exist: {effective_path}.') sys.exit(1) - if os.path.isdir(args.scan_dir): + if os.path.isdir(effective_path): if not scanner.scan_folder_with_options( - args.scan_dir, + scan_root or args.scan_dir, args.dep, scanner.winnowing.file_map, args.dep_scope, args.dep_scope_inc, args.dep_scope_exc, + filter_path=relative_target, ): sys.exit(1) - elif os.path.isfile(args.scan_dir): + elif os.path.isfile(effective_path): if not scanner.scan_file_with_options( - args.scan_dir, + effective_path, args.dep, scanner.winnowing.file_map, args.dep_scope, args.dep_scope_inc, args.dep_scope_exc, + file_id=relative_target, ): sys.exit(1) else: - print_stderr(f'Error: Path specified is neither a file or a folder: {args.scan_dir}.') + print_stderr(f'Error: Path specified is neither a file or a folder: {effective_path}.') sys.exit(1) elif args.dep: if not args.dependencies_only: @@ -1769,40 +1789,34 @@ def dependency(parser, args): args: Namespace Parsed arguments """ - if not args.scan_loc and not args.container: + if not args.scan_dir and not args.container: print_stderr('Please specify a file/folder or container image') parser.parse_args([args.subparser, '-h']) sys.exit(1) # Workaround to return syft scan results converted to our dependency output format if args.container: + if getattr(args, 'scan_root', None): + print_stderr('ERROR: --scan-root is not supported with --container.') + sys.exit(1) args.scan_loc = args.container return container_scan(parser, args, only_interim_results=True) - if not os.path.exists(args.scan_loc): - print_stderr(f'Error: File or folder specified does not exist: {args.scan_loc}.') + validate_scan_root(args) + effective_scan_dir = os.path.join(args.scan_root, args.scan_dir) if args.scan_root else args.scan_dir + if not os.path.exists(effective_scan_dir): + print_stderr(f'Error: File or folder specified does not exist: {effective_scan_dir}.') sys.exit(1) if args.output: initialise_empty_file(args.output) - if args.settings and args.skip_settings_file: - print_stderr('ERROR: Cannot specify both --settings and --skip-file-settings options.') - sys.exit(1) - scanoss_settings = None - if not args.skip_settings_file: - scanoss_settings = ScanossSettings(debug=args.debug, trace=args.trace, quiet=args.quiet) - try: - scanoss_settings.load_json_file(args.settings, args.scan_loc) - except ScanossSettingsError as e: - print_stderr(f'Error: {e}') - sys.exit(1) - + scanoss_settings = get_scanoss_settings_from_args(args) sc_deps = ScancodeDeps( debug=args.debug, quiet=args.quiet, trace=args.trace, sc_command=args.sc_command, timeout=args.sc_timeout, scanoss_settings=scanoss_settings, ) if not sc_deps.get_dependencies( - what_to_scan=args.scan_loc, result_output=args.output + what_to_scan=effective_scan_dir, result_output=args.output ): sys.exit(1) return None @@ -2826,8 +2840,10 @@ def folder_hashing_scan(parser, args): parser.parse_args([args.subparser, '-h']) sys.exit(1) - if not os.path.exists(args.scan_dir) or not os.path.isdir(args.scan_dir): - print_stderr(f'ERROR: The specified directory {args.scan_dir} does not exist') + validate_scan_root(args) + effective_scan_dir = os.path.join(args.scan_root, args.scan_dir) if args.scan_root else args.scan_dir + if not os.path.exists(effective_scan_dir) or not os.path.isdir(effective_scan_dir): + print_stderr(f'ERROR: The specified directory {effective_scan_dir} does not exist') sys.exit(1) scanner_config = create_scanner_config_from_args(args) @@ -2837,7 +2853,7 @@ def folder_hashing_scan(parser, args): client = ScanossGrpc(**asdict(grpc_config)) scanner = ScannerHFH( - scan_dir=args.scan_dir, + scan_dir=effective_scan_dir, config=scanner_config, client=client, scanoss_settings=scanoss_settings, @@ -2867,21 +2883,23 @@ def folder_hash(parser, args): parser.parse_args([args.subparser, '-h']) sys.exit(1) - if not os.path.exists(args.scan_dir) or not os.path.isdir(args.scan_dir): - print_stderr(f'ERROR: The specified directory {args.scan_dir} does not exist') + validate_scan_root(args) + effective_scan_dir = os.path.join(args.scan_root, args.scan_dir) if args.scan_root else args.scan_dir + if not os.path.exists(effective_scan_dir) or not os.path.isdir(effective_scan_dir): + print_stderr(f'ERROR: The specified directory {effective_scan_dir} does not exist') sys.exit(1) folder_hasher_config = create_folder_hasher_config_from_args(args) scanoss_settings = get_scanoss_settings_from_args(args) folder_hasher = FolderHasher( - scan_dir=args.scan_dir, + scan_dir=effective_scan_dir, config=folder_hasher_config, scanoss_settings=scanoss_settings, depth=args.depth, ) - folder_hasher.hash_directory(args.scan_dir) + folder_hasher.hash_directory(effective_scan_dir) folder_hasher.present(output_file=args.output, output_format=args.format) except Exception as e: print_stderr(f'ERROR: {e}') @@ -2924,15 +2942,50 @@ def container_scan(parser, args, only_interim_results: bool = False): sys.exit(1) +def validate_scan_root(args): + """Validate --scan-root and its relationship to the scan target. Exits on error.""" + if not args.scan_root: + return + scan_root: str = args.scan_root + if not args.scan_dir: + print_stderr('ERROR: --scan-root requires a scan target (FILE/DIR).') + sys.exit(1) + if not os.path.isdir(scan_root): + print_stderr(f'ERROR: --scan-root must be an existing directory: {scan_root}') + sys.exit(1) + effective_target = os.path.join(scan_root, args.scan_dir) + if not os.path.exists(effective_target): + print_stderr(f'ERROR: Resolved scan target does not exist: {effective_target}') + sys.exit(1) + try: + Path(effective_target).resolve().relative_to(Path(scan_root).resolve()) + except ValueError: + print_stderr(f'ERROR: Scan target escapes scan root: {args.scan_dir}') + sys.exit(1) + + def get_scanoss_settings_from_args(args): - scanoss_settings = None - if not args.skip_settings_file: - scanoss_settings = ScanossSettings(debug=args.debug, trace=args.trace, quiet=args.quiet) - try: - scanoss_settings.load_json_file(args.settings, args.scan_dir).set_file_type('new').set_scan_type('identify') - except ScanossSettingsError as e: - print_stderr(f'Error: {e}') - sys.exit(1) + settings = getattr(args, 'settings', None) + skip_settings_file = getattr(args, 'skip_settings_file', False) + if settings and skip_settings_file: + print_stderr('ERROR: Cannot specify both --settings and --skip-file-settings options.') + sys.exit(1) + if skip_settings_file: + return None + settings_root = getattr(args, 'scan_root', None) or getattr(args, 'scan_dir', None) + scanoss_settings = ScanossSettings(debug=args.debug, trace=args.trace, quiet=args.quiet) + try: + identify = getattr(args, 'identify', None) + ignore = getattr(args, 'ignore', None) + if identify: + scanoss_settings.load_json_file(identify, settings_root).set_file_type('legacy').set_scan_type('identify') + elif ignore: + scanoss_settings.load_json_file(ignore, settings_root).set_file_type('legacy').set_scan_type('blacklist') + else: + scanoss_settings.load_json_file(settings, settings_root).set_file_type('new') + except ScanossSettingsError as e: + print_stderr(f'Error: {e}') + sys.exit(1) return scanoss_settings diff --git a/src/scanoss/file_filters.py b/src/scanoss/file_filters.py index 33595374..e0cc2300 100644 --- a/src/scanoss/file_filters.py +++ b/src/scanoss/file_filters.py @@ -308,27 +308,29 @@ def __init__(self, debug: bool = False, trace: bool = False, quiet: bool = False self.file_folder_pat_spec = self._get_file_folder_pattern_spec(kwargs.get('operation_type', 'scanning')) self.size_pat_rules = self._get_size_limit_pattern_rules(kwargs.get('operation_type', 'scanning')) - def get_filtered_files_from_folder(self, root: str) -> List[str]: + @staticmethod + def _filter_parts(path_value: str) -> tuple: + return tuple(part for part in Path(path_value).parts if part not in ('.', '')) + + def get_filtered_files_from_folder(self, root: str, filter_path: str = None) -> List[str]: """ - Retrieve a list of files to scan or fingerprint from a given directory root based on filter settings. + Filters and retrieves a list of files from a specified root directory based on various optional conditions. - Args: - root (str): Root directory to scan or fingerprint + This method performs a recursive directory traversal starting from the provided root directory, + collects all file paths, and optionally filters the files or directories based on directory skip rules, + global skip rules, or a specific path filter. The filtered list of file paths is then returned. + + Parameters: + root: str + The root directory path to start the recursive search for files. + filter_path: str, optional + A specific filter path used to limit the selection of files to certain subdirectories. Default is None. Returns: - list[str]: Filtered list of files to scan or fingerprint + List[str] + A list of file paths, as strings, that meet the specified filtering criteria. """ - if self.debug: - if self.file_folder_pat_spec: - self.print_stderr(f'Running with {len(self.file_folder_pat_spec)} pattern filters.') - if self.size_pat_rules: - self.print_stderr(f'Running with {len(self.size_pat_rules)} size pattern rules.') - if self.skip_size: - self.print_stderr(f'Running with global skip size: {self.skip_size}') - if self.skip_extensions: - self.print_stderr(f'Running with extra global skip extensions: {self.skip_extensions}') - if self.skip_folders: - self.print_stderr(f'Running with extra global skip folders: {self.skip_folders}') + self._print_filtered_files_folder_debug() all_files = [] root_path = Path(root).resolve() if not root_path.exists() or not root_path.is_dir(): @@ -340,8 +342,7 @@ def get_filtered_files_from_folder(self, root: str) -> List[str]: rel_path = dir_path.relative_to(root_path) if dir_path.is_symlink(): # TODO should we skip symlink folders? self.print_msg(f'WARNING: Found symbolic link folder: {dir_path}') - - if self.should_skip_dir(str(rel_path)): # Current directory should be skipped + if self.should_skip_dir(str(rel_path), filter_path): # The current directory should be skipped dirnames.clear() continue for filename in filenames: @@ -349,7 +350,30 @@ def get_filtered_files_from_folder(self, root: str) -> List[str]: all_files.append(str(file_path)) # End os.walk loop # Now filter the files and return the reduced list - return self.get_filtered_files_from_files(all_files, str(root_path)) + files = self.get_filtered_files_from_files(all_files, str(root_path)) + if filter_path: + filter_parts = self._filter_parts(filter_path) + if filter_parts: + files = [f for f in files if tuple(Path(f).parts[:len(filter_parts)]) == filter_parts] + return files + + def _print_filtered_files_folder_debug(self): + """ + Print debug information regarding filtered files, folders, and rules when + debug mode is enabled. + """ + if self.debug: + if self.file_folder_pat_spec: + self.print_stderr(f'Running with {len(self.file_folder_pat_spec)} pattern filters.') + if self.size_pat_rules: + self.print_stderr(f'Running with {len(self.size_pat_rules)} size pattern rules.') + if self.skip_size: + self.print_stderr(f'Running with global skip size: {self.skip_size}') + if self.skip_extensions: + self.print_stderr(f'Running with extra global skip extensions: {self.skip_extensions}') + if self.skip_folders: + self.print_stderr(f'Running with extra global skip folders: {self.skip_folders}') + def get_filtered_files_from_files(self, files: List[str], scan_root: Optional[str] = None) -> List[str]: """ @@ -502,15 +526,16 @@ def _get_operation_size_limits(self, file_path: str = None) -> tuple: # End rules loop return min_size, max_size - def should_skip_dir(self, dir_rel_path: str) -> bool: # noqa: PLR0911 + def should_skip_dir(self, dir_rel_path: str, filter_path: str = None) -> bool: # noqa: PLR0911 """ Check if a directory should be skipped based on operation type and default rules. Args: dir_rel_path (str): Relative path to the directory + filter_path (str): Optional filter path to check if the directory is within it Returns: - bool: True if directory should be skipped, False otherwise + bool: True if the directory should be skipped, False otherwise """ dir_name = os.path.basename(dir_rel_path) dir_path = Path(dir_rel_path) @@ -521,6 +546,14 @@ def should_skip_dir(self, dir_rel_path: str) -> bool: # noqa: PLR0911 ): self.print_debug(f'Skipping directory: {dir_rel_path} (hidden directory)') return True + if filter_path and dir_rel_path != '.': + filter_parts = self._filter_parts(filter_path) + dir_parts = self._filter_parts(dir_rel_path) + in_filter = tuple(dir_parts[:len(filter_parts)]) == filter_parts + is_ancestor = tuple(filter_parts[:len(dir_parts)]) == dir_parts + if not in_filter and not is_ancestor: + self.print_debug(f'Skipping directory: {dir_rel_path} (not in filter path)') + return True if self.all_folders: return False dir_name_lower = dir_name.lower() diff --git a/src/scanoss/scanner.py b/src/scanoss/scanner.py index 1467b328..4611c07e 100644 --- a/src/scanoss/scanner.py +++ b/src/scanoss/scanner.py @@ -393,6 +393,7 @@ def scan_folder_with_options( # noqa: PLR0913 dep_scope: SCOPE = None, dep_scope_include: str = None, dep_scope_exclude: str = None, + filter_path: str = None, ) -> bool: """ Scan the given folder for whatever scaning options that have been configured @@ -402,6 +403,7 @@ def scan_folder_with_options( # noqa: PLR0913 :param scan_dir: directory to scan :param deps_file: pre-parsed dependency file to decorate :param file_map: mapping of obfuscated files back into originals + :param filter_path: restrict scanning to this sub-path within scan_dir :return: True if successful, False otherwise """ @@ -416,8 +418,9 @@ def scan_folder_with_options( # noqa: PLR0913 if self.scan_output: self.print_msg(f'Writing results to {self.scan_output}...') if self.is_dependency_scan(): + dep_scan_dir = os.path.join(scan_dir, filter_path) if filter_path else scan_dir if not self.threaded_deps.run( - what_to_scan=scan_dir, + what_to_scan=dep_scan_dir, deps_file=deps_file, wait=False, dep_scope=dep_scope, @@ -426,14 +429,14 @@ def scan_folder_with_options( # noqa: PLR0913 ): # Kick off a background dependency scan success = False if self.is_file_or_snippet_scan(): - if not self.scan_folder(scan_dir): + if not self.scan_folder(scan_dir, filter_path): success = False if self.threaded_scan: if not self.__finish_scan_threaded(file_map): success = False return success - def scan_folder(self, scan_dir: str) -> bool: # noqa: PLR0912, PLR0915 + def scan_folder(self, scan_dir: str, filter_path: str = None) -> bool: # noqa: PLR0912, PLR0915 """ Scan the specified folder producing fingerprints, send to the SCANOSS API and return results @@ -473,7 +476,7 @@ def scan_folder(self, scan_dir: str) -> bool: # noqa: PLR0912, PLR0915 wfp_list = [] if self.wfp_output else None # Collect WFPs if output file is specified batch_context = None # Track SBOM context (purls, scan_type) for the current batch - to_scan_files = file_filters.get_filtered_files_from_folder(scan_dir) + to_scan_files = file_filters.get_filtered_files_from_folder(scan_dir, filter_path) for to_scan_file in to_scan_files: if self.threaded_scan and self.threaded_scan.stop_scanning(): self.print_stderr('Warning: Aborting fingerprinting as the scanning service is not available.') @@ -669,6 +672,7 @@ def scan_file_with_options( # noqa: PLR0913 dep_scope: SCOPE = None, dep_scope_include: str = None, dep_scope_exclude: str = None, + file_id: str = None, ) -> bool: """ Scan the given file for whatever scaning options that have been configured @@ -676,6 +680,7 @@ def scan_file_with_options( # noqa: PLR0913 :param file: file to scan :param deps_file: pre-parsed dependency file to decorate :param file_map: mapping of obfuscated files back into originals + :param file_id: override the file identifier used in WFP and path matching (defaults to file) :return: True if successful, False otherwise """ success = True @@ -699,20 +704,22 @@ def scan_file_with_options( # noqa: PLR0913 ): # Kick off a background dependency scan success = False if self.is_file_or_snippet_scan(): - if not self.scan_file(file): + if not self.scan_file(file, file_id): success = False if self.threaded_scan: if not self.__finish_scan_threaded(file_map): success = False return success - def scan_file(self, file: str) -> bool: + def scan_file(self, file: str, file_id: str = None) -> bool: """ Scan the specified file and produce a result Parameters ---------- file: str File to fingerprint and scan/identify + file_id: str + Override for WFP identifier and path matching (defaults to file) :return True if successful, False otherwise """ success = True @@ -721,11 +728,12 @@ def scan_file(self, file: str) -> bool: if not os.path.exists(file) or not os.path.isfile(file): raise Exception(f'ERROR: Specified files does not exist or is not a file: {file}') self.print_debug(f'Fingerprinting {file}...') - wfp = self.winnowing.wfp_for_file(file, file) + wfp_id = file_id or file + wfp = self.winnowing.wfp_for_file(file, wfp_id) if wfp is not None and wfp != '': if self.threaded_scan: file_context = ( - self.scanoss_settings.get_sbom_context(file) + self.scanoss_settings.get_sbom_context(wfp_id) if self.scanoss_settings else SbomContext.empty() ) sbom = file_context.to_payload() @@ -1090,7 +1098,7 @@ def wfp_contents(self, filename: str, contents: bytes, wfp_file: str = None): else: Scanner.print_stderr(f'Warning: No fingerprints generated for: {wfp_file}') - def wfp_file(self, scan_file: str, wfp_file: str = None): + def wfp_file(self, scan_file: str, wfp_file: str = None, file_id: str = None): """ Fingerprint the specified file """ @@ -1100,7 +1108,8 @@ def wfp_file(self, scan_file: str, wfp_file: str = None): raise Exception(f'ERROR: Specified file does not exist or is not a file: {scan_file}') self.print_debug(f'Fingerprinting {scan_file}...') - wfp = self.winnowing.wfp_for_file(scan_file, scan_file) + wfp_id = file_id or scan_file + wfp = self.winnowing.wfp_for_file(scan_file, wfp_id) if wfp: if wfp_file: self.print_stderr(f'Writing fingerprints to {wfp_file}') @@ -1111,7 +1120,7 @@ def wfp_file(self, scan_file: str, wfp_file: str = None): else: Scanner.print_stderr(f'Warning: No fingerprints generated for: {scan_file}') - def wfp_folder(self, scan_dir: str, wfp_file: str = None): + def wfp_folder(self, scan_dir: str, wfp_file: str = None, filter_path: str = None): """ Fingerprint the specified folder producing fingerprints """ @@ -1137,7 +1146,7 @@ def wfp_folder(self, scan_dir: str, wfp_file: str = None): spinner_ctx = Spinner('Fingerprinting ') if (not self.quiet and self.isatty) else nullcontext() with spinner_ctx as spinner: - to_fingerprint_files = file_filters.get_filtered_files_from_folder(scan_dir) + to_fingerprint_files = file_filters.get_filtered_files_from_folder(scan_dir, filter_path) for file in to_fingerprint_files: if spinner: spinner.next() diff --git a/src/scanoss/scanoss_settings.py b/src/scanoss/scanoss_settings.py index 7a39fbaa..8483d790 100644 --- a/src/scanoss/scanoss_settings.py +++ b/src/scanoss/scanoss_settings.py @@ -237,36 +237,55 @@ def load_json_file(self, filepath: Optional[str] = None, scan_root: Optional[str Load the scan settings file. If no filepath is provided, scanoss.json will be used as default. Args: - filepath (str): Path to the SCANOSS settings file + :param filepath: Path to the SCANOSS settings file + :param scan_root: Path to the scan root directory/file """ - if not filepath: - filepath = DEFAULT_SCANOSS_JSON_FILE - - filepath = Path(scan_root) / filepath if scan_root else Path(filepath) - - json_file = filepath.resolve() - - if filepath == DEFAULT_SCANOSS_JSON_FILE and not json_file.exists(): - self.print_debug(f'Default settings file "{filepath}" not found. Skipping...') + file_path = DEFAULT_SCANOSS_JSON_FILE + if filepath: + file_path = Path(filepath) + # Builds a prioritised candidate paths for settings file from scan root or cwd + candidates = [] + # Prepend the scan root directory to the filepath if it's not an absolute path and not a relative path + if not file_path.is_absolute() and (not filepath or not filepath.startswith('.')): + if scan_root: + scan_root_path = Path(scan_root) + if scan_root_path.is_file(): # Attempt to load from the parent directory of the file + candidates.append(Path(scan_root_path.parent / file_path)) + else: + candidates.append(Path(scan_root_path / file_path)) # Attempt to load from scan root directory + candidates.append(Path(Path.cwd() / file_path)) # Attempt to load from the current working directory + # Fallback to the defined filepath + candidates.append(file_path) + # Loads the first valid settings file after validation and schema check + self.print_debug(f'Searching for settings file from: {candidates}...') + json_file = None + for candidate in candidates: + js_file = candidate.resolve() + if js_file.exists(): + json_file = js_file + break + # End for loop + if json_file: + self.print_msg(f'Loading settings file {json_file}...') + result = validate_json_file(json_file) + if not result.is_valid: + if result.error_code in (JSON_ERROR_FILE_NOT_FOUND, JSON_ERROR_FILE_EMPTY): + self.print_msg( + f'WARNING: The supplied settings file "{filepath}" was not found or is empty. Skipping...' + ) + return self + else: + raise ScanossSettingsError(f'Problem with settings file. {result.error}') + try: + validate(result.data, self.schema) + except Exception as e: + raise ScanossSettingsError(f'Invalid settings file. {e}') from e + self.data = result.data + self.print_debug(f'Loaded scan settings from: {json_file}') return self - self.print_msg(f'Loading settings file {filepath}...') - - result = validate_json_file(json_file) - if not result.is_valid: - if result.error_code in (JSON_ERROR_FILE_NOT_FOUND, JSON_ERROR_FILE_EMPTY): - self.print_msg( - f'WARNING: The supplied settings file "{filepath}" was not found or is empty. Skipping...' - ) - return self - else: - raise ScanossSettingsError(f'Problem with settings file. {result.error}') - try: - validate(result.data, self.schema) - except Exception as e: - raise ScanossSettingsError(f'Invalid settings file. {e}') from e - self.data = result.data - self.print_debug(f'Loading scan settings from: {filepath}') + # Not valid settings file found. Skip + self.print_debug(f'Default settings file "{file_path}" not found. Skipping...') return self def set_file_type(self, file_type: str): diff --git a/src/scanoss/scanossapi.py b/src/scanoss/scanossapi.py index a6816e88..c45ae0b9 100644 --- a/src/scanoss/scanossapi.py +++ b/src/scanoss/scanossapi.py @@ -49,6 +49,45 @@ SCAN_ENDPOINT = '/scan/direct' # scan endpoint path SCANOSS_SCAN_URL = os.environ.get('SCANOSS_SCAN_URL') if os.environ.get('SCANOSS_SCAN_URL') else DEFAULT_URL SCANOSS_API_KEY = os.environ.get('SCANOSS_API_KEY') if os.environ.get('SCANOSS_API_KEY') else '' +DEFAULT_RETRY_AFTER = 5 # Default backoff (seconds) when a rate-limit response gives no hint +MAX_RETRY_AFTER = 60 # Cap on backoff (seconds) to avoid pathological waits + + +def parse_retry_after(response, default: int = DEFAULT_RETRY_AFTER, maximum: int = MAX_RETRY_AFTER) -> int: + """ + Determine how long to back off before retrying a rate-limited request. + + Reads the standard 'Retry-After' response header (interpreted as seconds); if it is + missing or non-numeric, falls back to the 'retry_after' field in the JSON body; if + neither is available, uses the supplied default. The result is clamped to [0, maximum]. + + :param response: requests Response object (may be None) + :param default: fallback delay in seconds when no hint is provided + :param maximum: upper bound for the returned delay in seconds + :return: delay in seconds (int) + """ + delay = None + if response is not None: + # Prefer the standard Retry-After header (seconds) + header_val = response.headers.get('Retry-After') if getattr(response, 'headers', None) else None + if header_val is not None: + try: + delay = int(float(str(header_val).strip())) + except (ValueError, TypeError): + delay = None + # Fall back to the retry_after field in the JSON body + if delay is None: + try: + body = response.json() + if isinstance(body, dict) and body.get('retry_after') is not None: + delay = int(float(body.get('retry_after'))) + except Exception: # noqa: BLE001 - defensive: any body/parse issue falls back to default + delay = None + if delay is None: + delay = default + if delay < 0: + delay = default + return min(delay, maximum) class ScanossApi(ScanossBase): @@ -242,16 +281,34 @@ def scan(self, wfp: str, context: str = None, scan_id: int = None, sbom: dict = else: self.print_stderr(f'Warning: No response received from {self.url}. Retrying...') time.sleep(5) - elif r.status_code == requests.codes.service_unavailable: # Service limits most likely reached - self.print_stderr( - f'ERROR: SCANOSS API rejected the scan request ({request_id}) due to ' - f'service limits being exceeded' - ) - self.print_stderr(f'ERROR: Details: {r.text.strip()}') - raise Exception( - f'ERROR: {r.status_code} - The SCANOSS API request ({request_id}) rejected ' - f'for {self.url} due to service limits being exceeded.' - ) + elif r.status_code in ( + requests.codes.too_many_requests, # 429 - rate limit + requests.codes.service_unavailable, # 503 - transient service unavailability + ): + # Both are potentially transient: back off (honouring Retry-After) and retry + # rather than aborting. Only 429 is a rate limit; a 503 is a generic outage + # (backend down, gateway circuit breaker) and must not be misreported as + # "service limits exceeded". + rate_limited = r.status_code == requests.codes.too_many_requests + if retry > self.retry_limit: # Exhausted retries, fail surfacing the real response + if rate_limited: + raise Exception( + f'ERROR: {r.status_code} - The SCANOSS API request ({request_id}) for ' + f'{self.url} was rejected due to service limits/rate limit being exceeded. ' + f'Server response: {r.text.strip()}' + ) + raise Exception( + f'ERROR: {r.status_code} - The SCANOSS API is currently unavailable ' + f'({request_id}) for {self.url}. Server response: {r.text.strip()}' + ) + else: + backoff = parse_retry_after(r) + reason = 'Rate limit exceeded' if rate_limited else 'Service unavailable' + self.print_stderr( + f'Warning: {reason} (HTTP {r.status_code}) from {self.url}: ' + f'{r.text.strip()}. Backing off for {backoff}s before retrying...' + ) + time.sleep(backoff) elif r.status_code >= requests.codes.bad_request: if retry > self.retry_limit: # No response retry_limit or more times, fail self.save_bad_req_wfp(scan_files, request_id, scan_id) diff --git a/tests/cli-test.py b/tests/cli-test.py index aee4d74e..958e061d 100644 --- a/tests/cli-test.py +++ b/tests/cli-test.py @@ -26,8 +26,9 @@ import shutil import tempfile import unittest +from types import SimpleNamespace -from scanoss.cli import load_files_from_file, process_req_headers +from scanoss.cli import get_scanoss_settings_from_args, load_files_from_file, process_req_headers class TestLoadFilesFromFile(unittest.TestCase): @@ -129,4 +130,28 @@ def test_header_argument_processor(self): # Test that the malformed header was not included self.assertNotIn('generic-header2', processed_headers, - "Malformed header without colon separator should not be included") \ No newline at end of file + "Malformed header without colon separator should not be included") + + +class TestGetScanossSettingsFromArgs(unittest.TestCase): + """Covers all commands that share this helper (scan, dependency, wfp, folder-scan, + folder-hash), since --settings/--skip-settings-file conflict handling previously + lived only inside scan() and silently regressed for the other commands.""" + + def _args(self, **overrides): + defaults = dict( + debug=False, trace=False, quiet=True, + settings=None, skip_settings_file=False, + scan_root=None, scan_dir=None, + ) + defaults.update(overrides) + return SimpleNamespace(**defaults) + + def test_rejects_settings_with_skip_settings_file(self): + args = self._args(settings='scanoss.json', skip_settings_file=True) + with self.assertRaises(SystemExit): + get_scanoss_settings_from_args(args) + + def test_skip_settings_file_alone_returns_none(self): + args = self._args(skip_settings_file=True) + self.assertIsNone(get_scanoss_settings_from_args(args)) \ No newline at end of file diff --git a/tests/scanossapi-test.py b/tests/scanossapi-test.py index e28e4556..f933552f 100644 --- a/tests/scanossapi-test.py +++ b/tests/scanossapi-test.py @@ -23,7 +23,24 @@ """ import unittest -from scanoss.scanossapi import ScanossApi +from unittest.mock import MagicMock, patch + +import requests + +from scanoss.scanossapi import DEFAULT_RETRY_AFTER, MAX_RETRY_AFTER, ScanossApi, parse_retry_after + + +def _mock_response(status_code, headers=None, json_body=None, text='{}'): + resp = MagicMock() + resp.status_code = status_code + resp.headers = headers if headers is not None else {} + resp.text = text + if json_body is not None: + resp.json.return_value = json_body + else: + resp.json.return_value = {} + return resp + class MyTestCase(unittest.TestCase): @@ -35,4 +52,117 @@ def test_scanoss_generic_headers(self): for key, value in scanoss_api.headers.items(): if key not in required_keys: valid_headers = False - self.assertTrue(valid_headers) \ No newline at end of file + self.assertTrue(valid_headers) + + +class RetryAfterParsingTestCase(unittest.TestCase): + + def test_header_takes_precedence(self): + resp = _mock_response(503, headers={'Retry-After': '7'}, json_body={'retry_after': 99}) + self.assertEqual(parse_retry_after(resp), 7) + + def test_falls_back_to_json_body(self): + resp = _mock_response(503, headers={}, json_body={'retry_after': 12}) + self.assertEqual(parse_retry_after(resp), 12) + + def test_default_when_no_hint(self): + resp = _mock_response(503, headers={}, json_body={}) + self.assertEqual(parse_retry_after(resp), DEFAULT_RETRY_AFTER) + + def test_non_numeric_header_falls_back(self): + resp = _mock_response(503, headers={'Retry-After': 'soon'}, json_body={'retry_after': 4}) + self.assertEqual(parse_retry_after(resp), 4) + + def test_non_numeric_everything_uses_default(self): + resp = _mock_response(503, headers={'Retry-After': 'soon'}, json_body={'retry_after': 'nope'}) + self.assertEqual(parse_retry_after(resp), DEFAULT_RETRY_AFTER) + + def test_caps_at_maximum(self): + resp = _mock_response(503, headers={'Retry-After': '99999'}) + self.assertEqual(parse_retry_after(resp), MAX_RETRY_AFTER) + + def test_negative_uses_default(self): + resp = _mock_response(503, headers={'Retry-After': '-3'}) + self.assertEqual(parse_retry_after(resp), DEFAULT_RETRY_AFTER) + + def test_none_response_uses_default(self): + self.assertEqual(parse_retry_after(None), DEFAULT_RETRY_AFTER) + + +class RateLimitRetryTestCase(unittest.TestCase): + + @patch('scanoss.scanossapi.time.sleep', return_value=None) + def test_429_retry_after_then_success(self, mock_sleep): + """The gateway now returns 429 for rate limits: retry, honour Retry-After, then succeed.""" + api = ScanossApi(retry=3, quiet=True) + rate_limited = _mock_response( + requests.codes.too_many_requests, + headers={'Retry-After': '2', 'X-Ratelimit-Burst-Capacity': '50'}, + json_body={'error': 'Rate limit exceeded', 'retry_after': 2, 'strategy': 'token_bucket'}, + text='{"error":"Rate limit exceeded"}', + ) + ok = _mock_response(200, json_body={'result': 'ok'}) + api.session = MagicMock() + api.session.post.side_effect = [rate_limited, ok] + + result = api.scan('dummy-wfp') + + self.assertEqual(result, {'result': 'ok'}) + self.assertEqual(api.session.post.call_count, 2) # retried instead of aborting + mock_sleep.assert_called_once_with(2) # honoured Retry-After header + + @patch('scanoss.scanossapi.time.sleep', return_value=None) + def test_429_honours_differing_retry_after_across_retries(self, mock_sleep): + """Each retry must honour its own Retry-After value, not a fixed backoff.""" + api = ScanossApi(retry=5, quiet=True) + first = _mock_response(requests.codes.too_many_requests, headers={'Retry-After': '2'}) + second = _mock_response(requests.codes.too_many_requests, headers={'Retry-After': '4'}) + ok = _mock_response(200, json_body={'result': 'ok'}) + api.session = MagicMock() + api.session.post.side_effect = [first, second, ok] + + result = api.scan('dummy-wfp') + + self.assertEqual(result, {'result': 'ok'}) + self.assertEqual([c.args[0] for c in mock_sleep.call_args_list], [2, 4]) + + @patch('scanoss.scanossapi.time.sleep', return_value=None) + def test_429_aborts_after_retries_exhausted_with_rate_limit_message(self, mock_sleep): + api = ScanossApi(retry=1, quiet=True) + body = '{"error":"Rate limit exceeded","message":"Retry in 1s.","retry_after":1}' + rate_limited = _mock_response( + requests.codes.too_many_requests, headers={'Retry-After': '1'}, text=body, + ) + api.session = MagicMock() + api.session.post.return_value = rate_limited + + with self.assertRaises(Exception) as ctx: + api.scan('dummy-wfp') + msg = str(ctx.exception) + self.assertIn('429', msg) + self.assertIn('rate limit', msg.lower()) + self.assertIn(body, msg) # the actual server response body is surfaced + # retry_limit=1 => one initial attempt + one retry = 2 posts + self.assertEqual(api.session.post.call_count, 2) + + @patch('scanoss.scanossapi.time.sleep', return_value=None) + def test_503_generic_outage_not_reported_as_rate_limit(self, mock_sleep): + """A generic 503 (circuit breaker) must surface status + body, NOT claim a rate limit.""" + api = ScanossApi(retry=1, quiet=True) + body = 'upstream connect error or disconnect/reset before headers' + outage = _mock_response(requests.codes.service_unavailable, text=body) + api.session = MagicMock() + api.session.post.return_value = outage + + with self.assertRaises(Exception) as ctx: + api.scan('dummy-wfp') + msg = str(ctx.exception) + self.assertIn('503', msg) + self.assertIn('currently unavailable', msg.lower()) + self.assertNotIn('rate limit', msg.lower()) + self.assertIn(body, msg) # the actual server response body is surfaced + self.assertEqual(api.session.post.call_count, 2) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/test_file_filters.py b/tests/test_file_filters.py index eb34face..4341aff1 100644 --- a/tests/test_file_filters.py +++ b/tests/test_file_filters.py @@ -323,5 +323,54 @@ def test_combined_all_modes(self): self.assertEqual(sorted(filtered_files), sorted(expected_files)) +class TestFilterPath(unittest.TestCase): + """Tests for filter_path support in get_filtered_files_from_folder.""" + + def setUp(self): + self.file_filters = FileFilters(debug=True, all_extensions=True, all_folders=True, + hidden_files_folders=True, operation_type='scanning') + self.test_dir = tempfile.mkdtemp() + + def tearDown(self): + shutil.rmtree(self.test_dir) + + def _create(self, paths): + for p in paths: + full = os.path.join(self.test_dir, p) + os.makedirs(os.path.dirname(full), exist_ok=True) + with open(full, 'w') as f: + f.write('int main(){}') # non-empty so FileFilters doesn't skip it + + def test_filter_path_returns_only_subfolder_files(self): + self._create(['sub/a.c', 'sub/b.c', 'other/c.c', 'root.c']) + result = self.file_filters.get_filtered_files_from_folder(self.test_dir, 'sub') + self.assertEqual(sorted(result), ['sub/a.c', 'sub/b.c']) + + def test_filter_path_excludes_partial_name_match(self): + self._create(['sub/a.c', 'sub2/b.c']) + result = self.file_filters.get_filtered_files_from_folder(self.test_dir, 'sub') + self.assertEqual(result, ['sub/a.c']) + + def test_filter_path_allows_nested_subdir(self): + self._create(['sub/deep/a.c', 'sub/b.c', 'other/c.c']) + result = self.file_filters.get_filtered_files_from_folder(self.test_dir, 'sub') + self.assertEqual(sorted(result), ['sub/b.c', 'sub/deep/a.c']) + + def test_filter_path_nested_target(self): + self._create(['sub/deep/a.c', 'sub/b.c', 'other/c.c']) + result = self.file_filters.get_filtered_files_from_folder(self.test_dir, 'sub/deep') + self.assertEqual(result, ['sub/deep/a.c']) + + def test_filter_path_none_returns_all(self): + self._create(['sub/a.c', 'other/b.c', 'root.c']) + result = self.file_filters.get_filtered_files_from_folder(self.test_dir, None) + self.assertEqual(sorted(result), ['other/b.c', 'root.c', 'sub/a.c']) + + def test_filter_path_with_trailing_slash(self): + self._create(['sub/a.c', 'other/b.c']) + result = self.file_filters.get_filtered_files_from_folder(self.test_dir, 'sub/') + self.assertEqual(result, ['sub/a.c']) + + if __name__ == '__main__': unittest.main() diff --git a/tests/test_scanoss_settings.py b/tests/test_scanoss_settings.py new file mode 100644 index 00000000..0eee546c --- /dev/null +++ b/tests/test_scanoss_settings.py @@ -0,0 +1,152 @@ +""" +SPDX-License-Identifier: MIT + + Copyright (c) 2026, SCANOSS + + Permission is hereby granted, free of charge, to any person obtaining a copy + of this software and associated documentation files (the "Software"), to deal + in the Software without restriction, including without limitation the rights + to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + copies of the Software, and to permit persons to whom the Software is + furnished to do so, subject to the following conditions: + + The above copyright notice and this permission notice shall be included in + all copies or substantial portions of the Software. + + THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN + THE SOFTWARE. +""" + +import json +import os +import tempfile +import unittest +from pathlib import Path +from unittest.mock import MagicMock, patch + +from src.scanoss.scanner import Scanner +from src.scanoss.scanoss_settings import ScanossSettings +from src.scanoss.scantype import ScanType + + +class TestScanossSettingsLoadJsonFile(unittest.TestCase): + def setUp(self): + self.settings_payload = { + "settings": {}, + "bom": { + "include": [], + "exclude": [], + "remove": [], + "replace": [] + } + } + self.original_cwd = os.getcwd() + self.tests_dir = Path(__file__).resolve().parent + os.chdir(self.tests_dir) + + def tearDown(self): + os.chdir(self.original_cwd) + + def _write_settings_file(self, directory: Path, filename: str = "scanoss.json") -> Path: + file_path = directory / filename + file_path.write_text(json.dumps(self.settings_payload), encoding="utf-8") + return file_path + + def test_loads_settings_from_scan_root_folder_first(self): + with tempfile.TemporaryDirectory() as tmp_dir: + tmp_path = Path(tmp_dir) + scan_root = tmp_path / "project" + scan_root.mkdir() + + settings_file = self._write_settings_file(scan_root) + + settings = ScanossSettings(debug=True) + result = settings.load_json_file(None, str(scan_root)) + + self.assertIs(result, settings) + self.assertEqual(settings.data, self.settings_payload) + self.assertTrue(settings_file.exists()) + + def test_falls_back_to_cwd_when_missing_in_scan_root_folder(self): + with tempfile.TemporaryDirectory() as tmp_dir: + tmp_path = Path(tmp_dir) + scan_root = tmp_path / "project" + scan_root.mkdir() + + cwd_settings = tmp_path / "scanoss.json" + cwd_settings.write_text(json.dumps(self.settings_payload), encoding="utf-8") + + settings = ScanossSettings(debug=True) + with patch("os.getcwd", return_value=str(tmp_path)): + with patch("pathlib.Path.cwd", return_value=tmp_path): + result = settings.load_json_file(None, str(scan_root)) + + self.assertIs(result, settings) + self.assertEqual(settings.data, self.settings_payload) + + def test_loads_settings_from_parent_directory_for_file_target(self): + with tempfile.TemporaryDirectory() as tmp_dir: + tmp_path = Path(tmp_dir) + target_dir = tmp_path / "project" + target_dir.mkdir() + + target_file = target_dir / "foo.c" + target_file.write_text("int main() {}", encoding="utf-8") + + settings_file = self._write_settings_file(target_dir) + + settings = ScanossSettings() + result = settings.load_json_file(None, str(target_file)) + + self.assertIs(result, settings) + self.assertEqual(settings.data, self.settings_payload) + self.assertTrue(settings_file.exists()) + + def test_returns_empty_settings_when_no_file_found(self): + with tempfile.TemporaryDirectory() as tmp_dir: + tmp_path = Path(tmp_dir) + scan_root = tmp_path / "project" + scan_root.mkdir() + + settings = ScanossSettings(debug=True) + result = settings.load_json_file(None, str(scan_root)) + + self.assertIs(result, settings) + self.assertEqual(settings.data, {}) + + +class TestScanFolderWithOptionsDependencyScope(unittest.TestCase): + """Regression tests: with --scan-root , dependency scanning must be + scoped to / (via filter_path), matching fingerprinting's scope - + not the whole scan root, which would pull in dependencies from sibling paths.""" + + def _make_scanner(self): + scanner = Scanner( + quiet=True, + nb_threads=0, + scan_options=ScanType.SCAN_DEPENDENCIES.value, + ) + scanner.threaded_deps.run = MagicMock(return_value=True) + return scanner + + def test_dependency_scan_scoped_to_filter_path(self): + scanner = self._make_scanner() + scanner.scan_folder_with_options('.', filter_path='subdir') + + scanner.threaded_deps.run.assert_called_once() + self.assertEqual( + scanner.threaded_deps.run.call_args.kwargs['what_to_scan'], + os.path.join('.', 'subdir'), + ) + + def test_dependency_scan_uses_scan_dir_when_no_filter_path(self): + scanner = self._make_scanner() + scanner.scan_folder_with_options('.') + + scanner.threaded_deps.run.assert_called_once() + self.assertEqual(scanner.threaded_deps.run.call_args.kwargs['what_to_scan'], '.')