Fix Pipenv venv discovery settings view (#645, #546)

crates/socket-patch-cli/tests/in_process_python_envs.rs

@@ -546,6 +546,8 @@ const PIPENV_VARS: &[&str] = &[
     "PIPENV_NO_VENV_IN_PROJECT",
     "PIPENV_CUSTOM_VENV_NAME",
     "PIPENV_PIPFILE",
+    "PIPENV_DONT_LOAD_ENV",
+    "PIPENV_DOTENV_LOCATION",
 ];
 
 /// Run `scan` with exactly `env` set among [`PIPENV_VARS`].
@@ -612,33 +614,104 @@ async fn pipenv_opt_outs_keep_activated_virtual_env_from_hijacking_scan() {
     }
 }
 
-/// #334: Pipenv never uses `venv/`, and `PIPENV_VENV_IN_PROJECT=0` makes it
-/// ignore a `./.venv` directory, so neither may shadow Pipenv's venv.
+/// #334: Pipenv never uses `venv/`, so it may not shadow Pipenv's venv.
 #[tokio::test]
 #[serial]
 async fn pipenv_stray_venv_dirs_do_not_shadow_the_pipenv_venv() {
-    for (stray, opt_out) in [("venv", None), (".venv", Some("0"))] {
+    let (_tmp, project, workon) = pipenv_project();
+    let stray_site = venv_site_packages(&project.join("venv"), "python3.12");
+    std::fs::create_dir_all(&stray_site).unwrap();
+    write_dist_info(&stray_site, "stray_decoy", "6.6.6");
+    let server = MockServer::start().await;
+    mock_batch_empty(&server).await;
+    let env: Vec<(&str, &Path)> = vec![
+        ("WORKON_HOME", &workon),
+        ("PIPENV_CUSTOM_VENV_NAME", Path::new("proj-env")),
+    ];
+    let code = scan_with_pipenv_env(default_args(&project, server.uri()), &env).await;
+    assert_eq!(code, 0);
+    let bodies = batch_bodies(&server).await;
+    assert_discovered(&bodies, "pkg:pypi/pipenv-pkg@1.0.0");
+    assert_not_discovered(&bodies, "pkg:pypi/stray-decoy@6.6.6");
+}
+
+/// #645: with `PIPENV_VENV_IN_PROJECT=0` (or `PIPENV_NO_VENV_IN_PROJECT=1`)
+/// only Pipenv 2023.11.14+ ignores a `./.venv` directory; 2018.11 through
+/// 2023.10.24 still use it. Both venvs are scanned, so whichever one the
+/// installed Pipenv uses is patched.
+#[tokio::test]
+#[serial]
+async fn pipenv_explicit_not_in_project_still_scans_dot_venv() {
+    for (name, value) in [
+        ("PIPENV_VENV_IN_PROJECT", "0"),
+        ("PIPENV_NO_VENV_IN_PROJECT", "1"),
+    ] {
         let (_tmp, project, workon) = pipenv_project();
-        let stray_site = venv_site_packages(&project.join(stray), "python3.12");
-        std::fs::create_dir_all(&stray_site).unwrap();
-        write_dist_info(&stray_site, "stray_decoy", "6.6.6");
+        let dot_site = venv_site_packages(&project.join(".venv"), "python3.12");
+        std::fs::create_dir_all(&dot_site).unwrap();
+        write_dist_info(&dot_site, "dot_venv_pkg", "1.0.0");
         let server = MockServer::start().await;
         mock_batch_empty(&server).await;
-        let mut env: Vec<(&str, &Path)> = vec![
+        let env: Vec<(&str, &Path)> = vec![
             ("WORKON_HOME", &workon),
             ("PIPENV_CUSTOM_VENV_NAME", Path::new("proj-env")),
+            (name, Path::new(value)),
         ];
-        if let Some(value) = opt_out {
-            env.push(("PIPENV_VENV_IN_PROJECT", Path::new(value)));
-        }
         let code = scan_with_pipenv_env(default_args(&project, server.uri()), &env).await;
-        assert_eq!(code, 0, "{stray}");
+        assert_eq!(code, 0, "{name}={value}");
         let bodies = batch_bodies(&server).await;
         assert_discovered(&bodies, "pkg:pypi/pipenv-pkg@1.0.0");
-        assert_not_discovered(&bodies, "pkg:pypi/stray-decoy@6.6.6");
+        assert_discovered(&bodies, "pkg:pypi/dot-venv-pkg@1.0.0");
     }
 }
 
+/// #546: Pipenv loads the project's `.env` before it picks the venv, so a
+/// `PIPENV_CUSTOM_VENV_NAME` or `WORKON_HOME` there decides which venv is
+/// scanned (and `PIPENV_DONT_LOAD_ENV` turns that off).
+#[tokio::test]
+#[serial]
+async fn pipenv_dotenv_settings_pick_the_scanned_venv() {
+    // Name in .env, WORKON_HOME exported.
+    let (_tmp, project, workon) = pipenv_project();
+    std::fs::write(project.join(".env"), "PIPENV_CUSTOM_VENV_NAME=proj-env\n").unwrap();
+    let server = MockServer::start().await;
+    mock_batch_empty(&server).await;
+    let code = scan_with_pipenv_env(
+        default_args(&project, server.uri()),
+        &[("WORKON_HOME", &workon)],
+    )
+    .await;
+    assert_eq!(code, 0);
+    assert_discovered(&batch_bodies(&server).await, "pkg:pypi/pipenv-pkg@1.0.0");
+
+    // Both in .env, nothing exported.
+    std::fs::write(
+        project.join(".env"),
+        format!(
+            "export WORKON_HOME=\"{}\"\nPIPENV_CUSTOM_VENV_NAME=proj-env # named\n",
+            // python-dotenv decodes `\r`, `\t`... in double quotes.
+            workon.display().to_string().replace('\\', "/")
+        ),
+    )
+    .unwrap();
+    let server = MockServer::start().await;
+    mock_batch_empty(&server).await;
+    let code = scan_with_pipenv_env(default_args(&project, server.uri()), &[]).await;
+    assert_eq!(code, 0);
+    assert_discovered(&batch_bodies(&server).await, "pkg:pypi/pipenv-pkg@1.0.0");
+
+    // PIPENV_DONT_LOAD_ENV: Pipenv ignores .env, and so does discovery.
+    let server = MockServer::start().await;
+    mock_batch_empty(&server).await;
+    let code = scan_with_pipenv_env(
+        default_args(&project, server.uri()),
+        &[("PIPENV_DONT_LOAD_ENV", Path::new("1"))],
+    )
+    .await;
+    assert_eq!(code, 0);
+    assert_not_discovered(&batch_bodies(&server).await, "pkg:pypi/pipenv-pkg@1.0.0");
+}
+
 // ---------------------------------------------------------------------------
 // Package-manager-recorded envs: PDM's saved interpreter / PEP 582, and uv's
 // UV_PROJECT_ENVIRONMENT, ahead of a stray `./.venv` the manager never uses

crates/socket-patch-cli/tests/in_process_redirect_pipenv.rs

@@ -29,6 +29,8 @@
 
 use std::path::Path;
 
+#[path = "common/hermetic.rs"]
+mod hermetic;
 #[path = "vex_e2e_common/mod.rs"]
 mod vex_e2e_common;
 #[path = "vex_pipenv_pip_steps/mod.rs"]
@@ -602,3 +604,222 @@ async fn warm_venv_with_the_upstream_release_is_not_attested() {
     roll_back(tmp.path(), &server).await;
     assert_eq!(read(&lock_path), LOCK);
 }
+
+/// Native Pipenv resolves these .env settings before choosing its env.
+/// A healthy ambient interpreter or empty local env must not hide the
+/// selected stale installation from the hosted-byte/VEX check.
+#[tokio::test]
+#[serial]
+async fn dotenv_selected_pipenv_install_is_checked_before_vex() {
+    for case in [
+        "single-quoted",
+        "multiline",
+        "ignore-active",
+        "override-active",
+        "relative-active",
+    ] {
+        let server = MockServer::start().await;
+        mock_api(&server).await;
+        let tmp = tempfile::tempdir().unwrap();
+        let project = tmp.path().join("project");
+        std::fs::create_dir_all(&project).unwrap();
+        write_project_with_upstream_install(&project);
+        let workon = tmp.path().join("workon");
+        std::fs::create_dir_all(&workon).unwrap();
+        let actual = workon.join("actual");
+        std::fs::rename(project.join(".venv"), &actual).unwrap();
+        std::fs::create_dir_all(site_packages(&project)).unwrap();
+        let ambient_project = tmp.path().join("ambient-project");
+        std::fs::create_dir_all(&ambient_project).unwrap();
+        write_project_with_upstream_install(&ambient_project);
+        let ambient_file = site_packages(&ambient_project).join("urllib3/response.py");
+        std::fs::write(&ambient_file, PATCHED).unwrap();
+        let actual_file = if cfg!(windows) {
+            actual.join("Lib/site-packages/urllib3/response.py")
+        } else {
+            actual.join("lib/python3.12/site-packages/urllib3/response.py")
+        };
+        let dotenv = match case {
+            "single-quoted" => "NAME=actual\nPIPENV_CUSTOM_VENV_NAME='${NAME}'\n".to_string(),
+            "multiline" => "PIPENV_CUSTOM_VENV_NAME=actual\nAPP_SETTINGS=\"first\nPIPENV_CUSTOM_VENV_NAME=decoy\nlast\"\n".to_string(),
+            "ignore-active" => "PIPENV_IGNORE_VIRTUALENVS=1\nPIPENV_CUSTOM_VENV_NAME=actual\n".to_string(),
+            "relative-active" => "VIRTUAL_ENV=../workon/actual\n".to_string(),
+            _ => format!("VIRTUAL_ENV='{}'\n", actual.to_string_lossy().replace('\\', "/")),
+        };
+        std::fs::write(project.join(".env"), &dotenv).unwrap();
+        let vex = project.join("out.vex.json");
+        let mut cmd = tokio::process::Command::from(hermetic::command(Path::new(env!(
+            "CARGO_BIN_EXE_socket-patch"
+        ))));
+        for (key, _) in std::env::vars_os() {
+            let key_text = key.to_string_lossy();
+            if key_text.starts_with("SOCKET_")
+                || key_text.starts_with("PIPENV_")
+                || matches!(key_text.as_ref(), "VIRTUAL_ENV" | "WORKON_HOME")
+            {
+                cmd.env_remove(key);
+            }
+        }
+        for key in [
+            "SOCKET_OFFLINE",
+            "SOCKET_DEBUG",
+            "SOCKET_API_URL",
+            "SOCKET_PROXY_URL",
+        ] {
+            cmd.env_remove(key);
+        }
+        cmd.env("SOCKET_TELEMETRY_DISABLED", "1")
+            .env(MAJOR_ENV, "2026")
+            .env("WORKON_HOME", &workon)
+            .args(["scan", "--mode", "hosted", "--yes", "--json", "--cwd"])
+            .arg(&project)
+            .args([
+                "--api-url",
+                &server.uri(),
+                "--org",
+                ORG,
+                "--api-token",
+                "fake",
+                "--vex",
+            ])
+            .arg(&vex)
+            .args(["--vex-product", VEX_PRODUCT]);
+        if case.ends_with("active") {
+            cmd.env("VIRTUAL_ENV", ambient_project.join(".venv"));
+        }
+        let out = cmd.output().await.unwrap();
+        let json: serde_json::Value = serde_json::from_slice(&out.stdout).unwrap_or_else(|error| {
+            panic!(
+                "{case}: {error}: stdout={} stderr={}",
+                String::from_utf8_lossy(&out.stdout),
+                String::from_utf8_lossy(&out.stderr)
+            )
+        });
+        assert_eq!(out.status.code(), Some(1), "{case}: {json}");
+        assert!(
+            json["redirect"]["warnings"]
+                .as_array()
+                .unwrap()
+                .iter()
+                .any(|warning| warning["code"] == "redirect_pypi_stale_install"),
+            "{case}: {json}"
+        );
+        assert!(!vex.exists(), "{case}: stale bytes must not produce VEX");
+        assert_eq!(
+            std::fs::read(&actual_file).unwrap(),
+            UPSTREAM,
+            "the probe is read-only"
+        );
+        assert_eq!(
+            std::fs::read(&ambient_file).unwrap(),
+            PATCHED,
+            "the ambient env is unchanged"
+        );
+        assert_eq!(read(&project.join("Pipfile")), PIPFILE);
+        assert!(read(&project.join("Pipfile.lock")).contains(HOSTED_URL));
+    }
+}
+
+/// Pipenv 2018 shell can select a third dotenv WORKON_HOME while current
+/// Pipenv and pre-dotenv commands use a healthy cached environment.
+#[tokio::test]
+#[serial]
+async fn legacy_dotenv_workon_install_is_checked_before_vex() {
+    for forward_reference in [true, false] {
+        let server = MockServer::start().await;
+        mock_api(&server).await;
+        let tmp = tempfile::tempdir().unwrap();
+        let project = tmp.path().join("project");
+        let legacy = tmp.path().join("legacy-workon");
+        let cached = tmp.path().join("cached-workon");
+        let make_install = |seed: &Path, root: &Path, bytes: &[u8]| {
+            std::fs::create_dir_all(seed).unwrap();
+            std::fs::create_dir_all(root.parent().unwrap()).unwrap();
+            write_project_with_upstream_install(seed);
+            let relative = site_packages(seed)
+                .strip_prefix(seed.join(".venv"))
+                .unwrap()
+                .to_path_buf();
+            std::fs::rename(seed.join(".venv"), root).unwrap();
+            let file = root.join(relative).join("urllib3/response.py");
+            std::fs::write(&file, bytes).unwrap();
+            file
+        };
+        let stale_file = make_install(&project, &legacy.join("env"), UPSTREAM);
+        let healthy_file = make_install(&tmp.path().join("seed"), &cached.join("env"), PATCHED);
+        // The .venv file names a project-specific venv within WORKON_HOME
+        // in both native generations, avoiding unrelated directory scans.
+        std::fs::write(project.join(".venv"), "env\n").unwrap();
+        let legacy_text = legacy.to_string_lossy().replace('\\', "/");
+        let cached_text = cached.to_string_lossy().replace('\\', "/");
+        let dotenv = if forward_reference {
+            format!("WORKON_HOME=${{BASE}}\nBASE={legacy_text}\n")
+        } else {
+            format!("BASE={cached_text}\nWORKON_HOME=${{BASE}}\n")
+        };
+        std::fs::write(project.join(".env"), dotenv).unwrap();
+        let vex = project.join("out.vex.json");
+        let mut cmd = tokio::process::Command::from(hermetic::command(Path::new(env!(
+            "CARGO_BIN_EXE_socket-patch"
+        ))));
+        for (key, _) in std::env::vars_os() {
+            let text = key.to_string_lossy();
+            if text.starts_with("SOCKET_")
+                || text.starts_with("PIPENV_")
+                || matches!(text.as_ref(), "VIRTUAL_ENV" | "WORKON_HOME" | "BASE")
+            {
+                cmd.env_remove(key);
+            }
+        }
+        cmd.env("SOCKET_TELEMETRY_DISABLED", "1")
+            .env(MAJOR_ENV, "2026")
+            .env("HOME", tmp.path().join("home"))
+            .env("USERPROFILE", tmp.path().join("home"))
+            .env("WORKON_HOME", &cached)
+            .args(["scan", "--mode", "hosted", "--yes", "--json", "--cwd"])
+            .arg(&project)
+            .args([
+                "--api-url",
+                &server.uri(),
+                "--org",
+                ORG,
+                "--api-token",
+                "fake",
+                "--vex",
+            ])
+            .arg(&vex)
+            .args(["--vex-product", VEX_PRODUCT]);
+        if !forward_reference {
+            cmd.env("BASE", &legacy);
+        }
+        let out = cmd.output().await.unwrap();
+        let json: serde_json::Value = serde_json::from_slice(&out.stdout).unwrap_or_else(|error| {
+            panic!(
+                "{error}: stdout={} stderr={}",
+                String::from_utf8_lossy(&out.stdout),
+                String::from_utf8_lossy(&out.stderr)
+            )
+        });
+        assert_eq!(
+            out.status.code(),
+            Some(1),
+            "forward={forward_reference}: {json}"
+        );
+        assert!(
+            json["redirect"]["warnings"]
+                .as_array()
+                .unwrap()
+                .iter()
+                .any(|warning| warning["code"] == "redirect_pypi_stale_install"),
+            "{json}"
+        );
+        assert!(
+            !vex.exists(),
+            "a healthy cached copy cannot attest the stale shell copy"
+        );
+        assert_eq!(std::fs::read(stale_file).unwrap(), UPSTREAM);
+        assert_eq!(std::fs::read(healthy_file).unwrap(), PATCHED);
+        assert_eq!(read(&project.join(".venv")), "env\n");
+        assert_eq!(read(&project.join("Pipfile")), PIPFILE);
+    }
+}