audit-labs/audit-tools
A collection of scripts, queries, and other goodies you can use in an audit.
clone: git clone https://gitbay.org/audit-labs/audit-tools.git
e2ef7a5e2e8604c0d10c696b5f7c89c63bf803ee
verified · cmc
author: Christian Cleberg <hello@cleberg.net> · 2026-08-07T03:04:22Z
applications/aws/aws_iam_users.sh | 24 +++++++-------- applications/aws/aws_s3_buckets.sh | 24 +++++++-------- applications/github/collectors/members.py | 4 +-- applications/gitlab/collectors/approvals.py | 2 +- .../gitlab/collectors/branch_protections.py | 2 +- applications/gitlab/collectors/members.py | 2 +- applications/gitlab/collectors/pipelines.py | 2 +- applications/gitlab/collectors/projects.py | 2 +- databases/sql/passwords/passwords.py | 25 +++++++++------- os/linux/passwords.sh | 8 +++-- os/linux/report/linux.sh | 13 ++++++-- os/linux/ssh_root_login.sh | 12 ++++---- sampling/sample.html | 10 +++---- sampling/sampling_tool/cli.py | 35 +++++++++++----------- sampling/tests/test_validation.py | 3 +- tui/tests/test_aws_runner.py | 3 +- 16 files changed, 93 insertions(+), 78 deletions(-) @@ -9,7 +9,7 @@ ACCOUNT_NAME="" # --- Prerequisite check --- if ! command -v aws &> /dev/null || ! command -v jq &> /dev/null; then - echo "Error: Both AWS CLI and jq are required. Please install them and ensure they are in your PATH." + echo "Error: Both AWS CLI and jq are required. Please install them and ensure they are in your PATH." >&2 exit 1 fi @@ -19,15 +19,15 @@ echo "Fetching IAM Identity Center and Account details..." INSTANCE_ARN=$(aws sso-admin list-instances --query "Instances[0].InstanceArn" --output text) IDENTITY_STORE_ID=$(aws sso-admin list-instances --query "Instances[0].IdentityStoreId" --output text) -if [ -z "$INSTANCE_ARN" ] || [ -z "$IDENTITY_STORE_ID" ]; then - echo "Error: Could not find IAM Identity Center instance ARN or Identity Store ID." +if [[ -z "$INSTANCE_ARN" ]] || [[ -z "$IDENTITY_STORE_ID" ]]; then + echo "Error: Could not find IAM Identity Center instance ARN or Identity Store ID." >&2 exit 1 fi ACCOUNT_ID=$(aws organizations list-accounts --query "Accounts[?Name=='$ACCOUNT_NAME' && Status=='ACTIVE'].Id" --output text) -if [ -z "$ACCOUNT_ID" ]; then - echo "Error: Could not find an active AWS account with the name '$ACCOUNT_NAME'." +if [[ -z "$ACCOUNT_ID" ]]; then + echo "Error: Could not find an active AWS account with the name '$ACCOUNT_NAME'." >&2 exit 1 fi @@ -41,7 +41,7 @@ PROVISIONED_SETS_ARN=$(aws sso-admin list-permission-sets-provisioned-to-account --account-id "$ACCOUNT_ID" \ --query "PermissionSets[]" --output text) -if [ -z "$PROVISIONED_SETS_ARN" ]; then +if [[ -z "$PROVISIONED_SETS_ARN" ]]; then echo "No permission sets are provisioned for account '$ACCOUNT_NAME'." exit 0 fi @@ -64,14 +64,14 @@ for PS_ARN in $PROVISIONED_SETS_ARN; do --permission-set-arn "$PS_ARN" \ --query "AccountAssignments[]" --output json) - if [ "$(echo "$ACCOUNT_ASSIGNMENTS" | jq 'length')" -eq 0 ]; then + if [[ "$(echo "$ACCOUNT_ASSIGNMENTS" | jq 'length')" -eq 0 ]]; then echo " -> Permission Set ARN $PS_ARN is provisioned but has no active assignments." continue fi # Since there are assignments, let's get the permission set's details (policies, name) # Using a cache to avoid redundant calls if a PS is somehow listed twice - if [ -z "${PERMISSION_SET_CACHE[$PS_ARN]}" ]; then + if [[ -z "${PERMISSION_SET_CACHE[$PS_ARN]}" ]]; then echo " -> Fetching policies for Permission Set: $PS_ARN" PS_NAME=$(aws sso-admin describe-permission-set --instance-arn "$INSTANCE_ARN" --permission-set-arn "$PS_ARN" --query "PermissionSet.Name" --output text) MANAGED_POLICIES=$(aws sso-admin list-managed-policies-in-permission-set --instance-arn "$INSTANCE_ARN" --permission-set-arn "$PS_ARN" --query "AttachedManagedPolicies[].Arn" --output json) @@ -87,16 +87,16 @@ for PS_ARN in $PROVISIONED_SETS_ARN; do # Now process each assignment found for this permission set for row in $(echo "${ACCOUNT_ASSIGNMENTS}" | jq -r '.[] | @base64'); do - _jq() { echo ${row} | base64 --decode | jq -r ${1}; } + _jq() { echo ${row} | base64 --decode | jq -r ${1}; return 0; } PRINCIPAL_TYPE=$(_jq '.PrincipalType') PRINCIPAL_ID=$(_jq '.PrincipalId') # Get Principal (User/Group) Name, using a cache - if [ -z "${PRINCIPAL_NAME_CACHE[$PRINCIPAL_ID]}" ]; then + if [[ -z "${PRINCIPAL_NAME_CACHE[$PRINCIPAL_ID]}" ]]; then PRINCIPAL_NAME="" - if [ "$PRINCIPAL_TYPE" == "USER" ]; then + if [[ "$PRINCIPAL_TYPE" == "USER" ]]; then PRINCIPAL_NAME=$(aws identitystore describe-user --identity-store-id "$IDENTITY_STORE_ID" --user-id "$PRINCIPAL_ID" --query "UserName" --output text 2>/dev/null) - elif [ "$PRINCIPAL_TYPE" == "GROUP" ]; then + elif [[ "$PRINCIPAL_TYPE" == "GROUP" ]]; then PRINCIPAL_NAME=$(aws identitystore describe-group --identity-store-id "$IDENTITY_STORE_ID" --group-id "$PRINCIPAL_ID" --query "DisplayName" --output text 2>/dev/null) fi PRINCIPAL_NAME_CACHE[$PRINCIPAL_ID]=${PRINCIPAL_NAME:-"ID: $PRINCIPAL_ID"} @@ -18,7 +18,7 @@ echo "---" echo "1. Retrieving all bucket names..." BUCKET_LIST=$(aws s3api list-buckets --region "$MASTER_REGION" --query 'Buckets[].Name' --output text) -if [ -z "$BUCKET_LIST" ]; then +if [[ -z "$BUCKET_LIST" ]]; then echo "✅ No S3 buckets found in this account." exit 0 fi @@ -32,14 +32,14 @@ for BUCKET_NAME in $BUCKET_LIST; do # 2. Find the bucket region for REGION in $AWS_REGIONS; do BUCKET_LOCATION_RESPONSE=$(aws s3api get-bucket-location --bucket "$BUCKET_NAME" --region "$REGION" 2>/dev/null) - if [ $? -eq 0 ]; then + if [[ $? -eq 0 ]]; then LOCATION_CONSTRAINT=$(echo "$BUCKET_LOCATION_RESPONSE" | jq -r '.LocationConstraint') BUCKET_REGION=${LOCATION_CONSTRAINT:-"us-east-1"} break fi done - if [ -z "$BUCKET_REGION" ]; then + if [[ -z "$BUCKET_REGION" ]]; then echo " ⚠️ WARNING: Could not determine region for $BUCKET_NAME. Skipping all checks." echo "$BUCKET_NAME,UNKNOWN,N/A,N/A,N/A,N/A,UNKNOWN" >> "$REPORT_FILE" continue @@ -57,14 +57,14 @@ for BUCKET_NAME in $BUCKET_LIST; do # --- CHECK A: Public Access Block (PAB) --- PAB_STATUS=$(aws s3api get-public-access-block --bucket "$BUCKET_NAME" --region "$BUCKET_REGION" 2>/dev/null) - if [ $? -ne 0 ]; then + if [[ $? -ne 0 ]]; then # PAB Missing is the highest risk state. PAB_FULLY_RESTRICTED="CRITICAL-MISSING" OVERALL_PUBLIC_STATUS="TRUE - PAB Missing" else # Check if ALL four PAB flags are true PAB_CONFIG=$(echo "$PAB_STATUS" | jq -r '.PublicAccessBlockConfiguration') - if [ "$(echo "$PAB_CONFIG" | jq -r '.BlockPublicAcls and .IgnorePublicAcls and .BlockPublicPolicy and .RestrictPublicBuckets')" = "true" ]; then + if [[ "$(echo "$PAB_CONFIG" | jq -r '.BlockPublicAcls and .IgnorePublicAcls and .BlockPublicPolicy and .RestrictPublicBuckets')" = "true" ]]; then PAB_FULLY_RESTRICTED="TRUE" else PAB_FULLY_RESTRICTED="FALSE-VULNERABLE" @@ -74,9 +74,9 @@ for BUCKET_NAME in $BUCKET_LIST; do # --- CHECK B: Bucket Policy Status (If S3 service thinks it's public) --- POLICY_STATUS=$(aws s3api get-bucket-policy-status --bucket "$BUCKET_NAME" --region "$BUCKET_REGION" 2>/dev/null) - if [ $? -eq 0 ]; then + if [[ $? -eq 0 ]]; then POLICY_IS_PUBLIC=$(echo "$POLICY_STATUS" | jq -r '.PolicyStatus.IsPublic') - if [ "$POLICY_IS_PUBLIC" = "true" ]; then + if [[ "$POLICY_IS_PUBLIC" = "true" ]]; then OVERALL_PUBLIC_STATUS="TRUE - Policy" fi else @@ -87,13 +87,13 @@ for BUCKET_NAME in $BUCKET_LIST; do # --- CHECK C: Bucket ACLs (for AllUsers group) --- ACL_RESPONSE=$(aws s3api get-bucket-acl --bucket "$BUCKET_NAME" --region "$BUCKET_REGION" 2>/dev/null) - if [ $? -eq 0 ]; then + if [[ $? -eq 0 ]]; then # Find if any grant to 'http://acs.amazonaws.com/groups/global/AllUsers' exists # Check for READ access if echo "$ACL_RESPONSE" | jq -e '.Grants[] | select(.Grantee.URI=="http://acs.amazonaws.com/groups/global/AllUsers") | select(.Permission | test("READ|FULL_CONTROL"))' >/dev/null; then ACL_ALL_USERS_READ="TRUE" - if [ "$OVERALL_PUBLIC_STATUS" = "FALSE" ]; then + if [[ "$OVERALL_PUBLIC_STATUS" = "FALSE" ]]; then OVERALL_PUBLIC_STATUS="TRUE - ACL Read" fi fi @@ -101,7 +101,7 @@ for BUCKET_NAME in $BUCKET_LIST; do # Check for WRITE access (often less common for public, but still public exposure) if echo "$ACL_RESPONSE" | jq -e '.Grants[] | select(.Grantee.URI=="http://acs.amazonaws.com/groups/global/AllUsers") | select(.Permission | test("WRITE|FULL_CONTROL"))' >/dev/null; then ACL_ALL_USERS_WRITE="TRUE" - if [ "$OVERALL_PUBLIC_STATUS" = "FALSE" ]; then + if [[ "$OVERALL_PUBLIC_STATUS" = "FALSE" ]]; then OVERALL_PUBLIC_STATUS="TRUE - ACL Write" fi fi @@ -111,9 +111,9 @@ for BUCKET_NAME in $BUCKET_LIST; do fi # Final check for PAB failure (PAB is the highest authority) - if [ "$PAB_FULLY_RESTRICTED" = "CRITICAL-MISSING" ]; then + if [[ "$PAB_FULLY_RESTRICTED" = "CRITICAL-MISSING" ]]; then OVERALL_PUBLIC_STATUS="TRUE - PAB Missing (CRITICAL)" - elif [ "$OVERALL_PUBLIC_STATUS" != "FALSE" ] && [ "$PAB_FULLY_RESTRICTED" != "TRUE" ]; then + elif [[ "$OVERALL_PUBLIC_STATUS" != "FALSE" ]] && [[ "$PAB_FULLY_RESTRICTED" != "TRUE" ]]; then # If the bucket is found public by Policy or ACL AND PAB isn't fully set, confirm it's public : # Status already set by Policy or ACL check above fi @@ -132,7 +132,7 @@ def outside_collaborators(org, cfg, repo_collabs): return rows -def privileged_access(org, cfg, repo_collabs): +def privileged_access(_org, _cfg, repo_collabs): """ All users with admin permission on any repo. Accepts pre-fetched repo_collabs from fetch_repo_collaborators(). @@ -196,7 +196,7 @@ def team_permissions(org, cfg): return rows -def permission_matrix(org, cfg, repo_collabs): +def permission_matrix(_org, _cfg, repo_collabs): """ Full per-repo/per-user permission cross-reference. Accepts pre-fetched repo_collabs from fetch_repo_collaborators(). @@ -12,7 +12,7 @@ import requests from .api import paginate -def approval_rules(group, cfg, projects): +def approval_rules(_group, cfg, projects): rows = [] for p in projects: try: @@ -12,7 +12,7 @@ def _levels(entries): return ", ".join(e.get("access_level_description", "") for e in entries) or "(none)" -def branch_protections(group, cfg, projects): +def branch_protections(_group, cfg, projects): rows = [] for p in projects: try: @@ -44,7 +44,7 @@ def group_members(group, cfg): ] -def project_members(group, cfg, projects): +def project_members(_group, cfg, projects): """Direct and inherited members of every project in the group.""" rows = [] for p in projects: @@ -7,7 +7,7 @@ import requests from .api import paginate -def pipelines(group, cfg, projects): +def pipelines(_group, cfg, projects): rows = [] for p in projects: try: @@ -17,7 +17,7 @@ def fetch_projects(group, cfg): ) -def project_list(group, cfg, projects): +def project_list(_group, _cfg, projects): """Format the project cache into audit rows.""" return [ { @@ -5,6 +5,11 @@ Checks SQL Server user data for compliance with Windows policies. # Import packages import pandas as pd +# Report column labels (defined once to avoid duplicated string literals). +TYPE_CHECK = "Type Check" +POLICY_CHECK = "Policy Check" +EXPIRATION_CHECK = "Expiration Check" + # Load the data into a pandas DataFrame df_input = pd.read_csv("./data.csv") @@ -24,38 +29,38 @@ def apply_rules_and_report(df): for _, row in df.iterrows(): result = { "Name": row["name"], - "Type Check": "", - "Policy Check": "", - "Expiration Check": "", + TYPE_CHECK: "", + POLICY_CHECK: "", + EXPIRATION_CHECK: "", "Reason": "", } # Check the type_desc if row["type_desc"] == "SQL_LOGIN": - result["Type Check"] = "SQL_LOGIN" + result[TYPE_CHECK] = "SQL_LOGIN" elif row["type_desc"] == "WINDOWS_LOGIN": - result["Type Check"] = "N/A" + result[TYPE_CHECK] = "N/A" result["Reason"] = "Refer to Windows password policy." else: - result["Type Check"] = "Manual Review" + result[TYPE_CHECK] = "Manual Review" result["Reason"] = "Reviewer to manually review." # Check if password policy is enforced if row["is_policy_checked"] == 1: - result["Policy Check"] = "PASS" + result[POLICY_CHECK] = "PASS" result["Reason"] += """Password policy is enforced. Reviewer to check the assigned policy.""" else: - result["Policy Check"] = "FAIL" + result[POLICY_CHECK] = "FAIL" result["Reason"] += "Password policy is not enforced." # Check if password expiration is enforced if row["is_expiration_checked"] == 1: - result["Expiration Check"] = "PASS" + result[EXPIRATION_CHECK] = "PASS" result["Reason"] += """Password expiration is enforced. Reviewer to check the expiration policy.""" else: - result["Expiration Check"] = "FAIL" + result[EXPIRATION_CHECK] = "FAIL" result["Reason"] += "Password expiration is not enforced." report.append(result) @@ -4,11 +4,11 @@ extract_password_params() { echo "Checking /etc/pam.d/system-auth for password parameters..." - if [ -f /etc/pam.d/system-auth ]; then + if [[ -f /etc/pam.d/system-auth ]]; then # Extract the line containing the password complexity parameters param_line=$(grep -E 'difok=.* minlen=.* dcredit=.* ocredit=.* ucredit=.* lcredit=.* minclass=.* maxsequence=.*' /etc/pam.d/system-auth) - if [ -n "$param_line" ]; then + if [[ -n "$param_line" ]]; then echo "Password complexity parameters found:" echo "$param_line" echo "" @@ -44,12 +44,13 @@ extract_password_params() { else echo "/etc/pam.d/system-auth file not found." fi + return 0 } # Function to analyze /etc/login.defs analyze_login_defs() { echo "Analyzing /etc/login.defs..." - if [ -f /etc/login.defs ]; then + if [[ -f /etc/login.defs ]]; then echo "Contents of /etc/login.defs:" cat /etc/login.defs echo "" @@ -61,6 +62,7 @@ analyze_login_defs() { else echo "/etc/login.defs file not found." fi + return 0 } # Main script execution @@ -6,10 +6,13 @@ TRIM_COMMENTS=false # Function to log section header log_section() { + local section_num="$1" + local section_title="$2" echo -e "\n\n" >> "$REPORT_FILE" echo "==========================================" >> "$REPORT_FILE" - echo "# SECTION $1: $2" >> "$REPORT_FILE" + echo "# SECTION $section_num: $section_title" >> "$REPORT_FILE" echo "==========================================" >> "$REPORT_FILE" + return 0 } # Function to log file content @@ -27,12 +30,16 @@ log_file_content() { else echo "File $FILE_PATH not found!" >> "$REPORT_FILE" fi + return 0 } # Function to log command output log_command_output() { - echo "## $1" >> "$REPORT_FILE" - $2 >> "$REPORT_FILE" 2>&1 + local label="$1" + local command="$2" + echo "## $label" >> "$REPORT_FILE" + $command >> "$REPORT_FILE" 2>&1 + return 0 } # Check for sudo privileges @@ -1,14 +1,14 @@ #!/bin/bash # Check if the script is being run as root -if [ "$EUID" -ne 0 ]; then - echo "Error: This script must be run as root or with sudo." +if [[ "$EUID" -ne 0 ]]; then + echo "Error: This script must be run as root or with sudo." >&2 exit 1 fi # Check if the sshd_config file exists -if [ ! -f /etc/ssh/sshd_config ]; then - echo "Error: /etc/ssh/sshd_config not found." +if [[ ! -f /etc/ssh/sshd_config ]]; then + echo "Error: /etc/ssh/sshd_config not found." >&2 exit 1 fi @@ -28,7 +28,7 @@ if ! echo "$permit_root_login" | grep -q "no"; then # Look for an explicitly set AuthorizedKeysFile path auth_keys_path_line=$(grep -E "^[[:space:]]*AuthorizedKeysFile" /etc/ssh/sshd_config) - if [ -n "$auth_keys_path_line" ]; then + if [[ -n "$auth_keys_path_line" ]]; then # An explicit path is set. Extract the path. # This removes the 'AuthorizedKeysFile' keyword and leading/trailing whitespace. auth_keys_path=$(echo "$auth_keys_path_line" | awk '{print $2}') @@ -45,7 +45,7 @@ if ! echo "$permit_root_login" | grep -q "no"; then fi echo "Checking for file at: $actual_path" - if [ -f "$actual_path" ]; then + if [[ -f "$actual_path" ]]; then echo "[CRITICAL] Found authorized keys file for root at $actual_path" echo "Contents:" echo "----------------------------------------" @@ -80,7 +80,7 @@ function handleFormSubmit(event) { // Use the custom seed if provided; otherwise draw a strong random seed and // write it back so the (reproducible) sample can always be tied to a seed. const seed = customSeedInput - ? parseInt(customSeedInput) + ? Number.parseInt(customSeedInput) : crypto.getRandomValues(new Uint32Array(1))[0] % 1000000; if (!customSeedInput) { document.getElementById('customSeed').value = seed; @@ -89,16 +89,16 @@ function handleFormSubmit(event) { } function generateSamples(seed) { - const populationSize = parseInt(document.getElementById('populationSize').value); - const sampleSize = parseInt(document.getElementById('sampleSize').value); - const replacementSize = parseInt(document.getElementById('replacementSize').value || 0); + const populationSize = Number.parseInt(document.getElementById('populationSize').value); + const sampleSize = Number.parseInt(document.getElementById('sampleSize').value); + const replacementSize = Number.parseInt(document.getElementById('replacementSize').value || 0); const resultsDiv = document.getElementById('results'); // Clear previous results resultsDiv.innerHTML = ''; // Validate inputs - if (isNaN(populationSize) || isNaN(sampleSize) || populationSize <= 0 || sampleSize <= 0) { + if (Number.isNaN(populationSize) || Number.isNaN(sampleSize) || populationSize <= 0 || sampleSize <= 0) { alert("Please enter valid numbers for required fields."); return; } @@ -23,6 +23,11 @@ from .reconciliation import build_reconciliation, build_strata_summary from .reporting import RunLogger, build_methodology from .validation import validate_and_prepare +# Output filenames, defined once so writer and tracker never drift. +_POPULATION_VALIDATED_CSV = "population_validated.csv" +_EXCLUDED_ROWS_CSV = "excluded_rows.csv" +_DUPLICATE_IDS_CSV = "duplicate_ids.csv" + def build_parser() -> ArgumentParser: parser = ArgumentParser(description="Generate documented audit samples.") @@ -211,7 +216,6 @@ def run(options) -> Path: _write_outputs( run_dir, options, - source, validated, excluded_rows, duplicate_rows, @@ -227,8 +231,6 @@ def run(options) -> Path: print(f"ERROR: {exc}", file=sys.stderr) _write_failure_outputs( run_dir, - options, - source, filtered, excluded_rows, duplicate_rows, @@ -307,7 +309,6 @@ def add_sample_metadata( def _write_outputs( run_dir: Path, options, - source: pd.DataFrame, validated: pd.DataFrame, excluded_rows: pd.DataFrame, duplicate_rows: pd.DataFrame, @@ -315,17 +316,17 @@ def _write_outputs( strata_rows: list[dict[str, object]], output_files: list[str], ) -> None: - write_csv(validated, run_dir / "population_validated.csv") - _track(output_files, "population_validated.csv") + write_csv(validated, run_dir / _POPULATION_VALIDATED_CSV) + _track(output_files, _POPULATION_VALIDATED_CSV) if options.method in {"random", "stratified"}: write_csv(sample, run_dir / "sample.csv") _track(output_files, "sample.csv") if not excluded_rows.empty: - write_csv(excluded_rows, run_dir / "excluded_rows.csv") - _track(output_files, "excluded_rows.csv") + write_csv(excluded_rows, run_dir / _EXCLUDED_ROWS_CSV) + _track(output_files, _EXCLUDED_ROWS_CSV) if not duplicate_rows.empty: - write_csv(duplicate_rows, run_dir / "duplicate_ids.csv") - _track(output_files, "duplicate_ids.csv") + write_csv(duplicate_rows, run_dir / _DUPLICATE_IDS_CSV) + _track(output_files, _DUPLICATE_IDS_CSV) if options.method == "stratified": write_csv(build_strata_summary(strata_rows), run_dir / "strata_summary.csv") _track(output_files, "strata_summary.csv") @@ -333,22 +334,20 @@ def _write_outputs( def _write_failure_outputs( run_dir: Path, - options, - source: pd.DataFrame, filtered: pd.DataFrame, excluded_rows: pd.DataFrame, duplicate_rows: pd.DataFrame, output_files: list[str], ) -> None: if not filtered.empty: - write_csv(filtered, run_dir / "population_validated.csv") - _track(output_files, "population_validated.csv") + write_csv(filtered, run_dir / _POPULATION_VALIDATED_CSV) + _track(output_files, _POPULATION_VALIDATED_CSV) if not excluded_rows.empty: - write_csv(excluded_rows, run_dir / "excluded_rows.csv") - _track(output_files, "excluded_rows.csv") + write_csv(excluded_rows, run_dir / _EXCLUDED_ROWS_CSV) + _track(output_files, _EXCLUDED_ROWS_CSV) if not duplicate_rows.empty: - write_csv(duplicate_rows, run_dir / "duplicate_ids.csv") - _track(output_files, "duplicate_ids.csv") + write_csv(duplicate_rows, run_dir / _DUPLICATE_IDS_CSV) + _track(output_files, _DUPLICATE_IDS_CSV) def _concat_nonempty(frames: list[pd.DataFrame]) -> pd.DataFrame: @@ -34,8 +34,9 @@ def test_duplicate_ids_fail_by_default_and_write_duplicate_file(tmp_path): source, index=False ) + options = _options(source, tmp_path / "out") with pytest.raises(AuditSamplingError): - run(_options(source, tmp_path / "out")) + run(options) run_dir = next((tmp_path / "out").glob("sample_*")) duplicates = pd.read_csv(run_dir / "duplicate_ids.csv") @@ -101,7 +101,8 @@ def test_session_build_failure_is_reported(tmp_path, fake_checks, monkeypatch): assert ("error", "AWS session") in kinds # Run still ends with a summary and writes the (empty) package. summary = [e for e in events if e.kind == "summary"] - assert summary and summary[0].count == 0 + assert summary + assert summary[0].count == 0 assert sections == [] assert os.path.exists(tmp_path / "summary.txt")