Python client example for keychain-auth integration

Build a Python client using Unix sockets on macOS/Linux with FD_CLOEXEC enforcement and Windows Named Pipes via pywin32 for cross-platform access.

This page covers a complete Python client that connects to the keychain-auth daemon over Unix domain sockets on macOS and Linux, and via Windows Named Pipes on Windows. You will learn how to enforce FD_CLOEXEC using fcntl, handle the pywin32 dependency on Windows, send newline-terminated JSON requests, and handle daemon error responses.

Full client implementation

The following is a complete Python client you can copy into your project and adapt with your service name and request payloads.

import os
import sys
import json
import socket

class KeychainAuthClient:
    def __init__(self):
        self.conn = None
        self._connect()

    def _connect(self):
        if sys.platform == "win32":
            # Connect to Windows Named Pipe
            self.pipe_path = r"\\.\pipe\keychain-auth"
            try:
                # Open pipe handle using native Win32 calls
                import win32file
                import win32pipe
                self.handle = win32file.CreateFile(
                    self.pipe_path,
                    win32file.GENERIC_READ | win32file.GENERIC_WRITE,
                    0, None, win32file.OPEN_EXISTING, 0, None
                )
            except ImportError:
                raise RuntimeError("Please install 'pywin32' package for named pipe connections on Windows.")
            except Exception as e:
                raise RuntimeError(f"keychain-auth daemon is not running: {e}")
        else:
            # Unix Domain Socket connection for macOS/Linux
            import platform
            if platform.system() == "Darwin":
                socket_path = os.path.expanduser("~/Library/Application Support/keychain-auth/agent.sock")
            else:
                runtime_dir = os.environ.get("XDG_RUNTIME_DIR", os.path.expanduser("~/.cache"))
                socket_path = os.path.join(runtime_dir, "keychain-auth", "agent.sock")
            
            try:
                # Create socket with standard Unix options
                self.conn = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
                
                # Enforce O_CLOEXEC on the socket descriptor to prevent leakage across fork/exec
                import fcntl
                flags = fcntl.fcntl(self.conn.fileno(), fcntl.F_GETFD)
                fcntl.fcntl(self.conn.fileno(), fcntl.F_SETFD, flags | fcntl.FD_CLOEXEC)
                
                self.conn.connect(socket_path)
            except Exception as e:
                raise RuntimeError(f"keychain-auth daemon is not running: {e}")

    def send_request(self, request_dict):
        payload = json.dumps(request_dict) + "\n"
        
        if sys.platform == "win32":
            # Write/Read Windows Named Pipe
            import win32file
            win32file.WriteFile(self.handle, payload.encode('utf-8'))
            
            # Read single-line JSON response
            response_bytes = b""
            while True:
                err, chunk = win32file.ReadFile(self.handle, 1024)
                response_bytes += chunk
                if b"\n" in response_bytes:
                    break
        else:
            # Write/Read Unix Domain Socket
            self.conn.sendall(payload.encode('utf-8'))
            
            # Read single-line JSON response
            response_bytes = b""
            while True:
                chunk = self.conn.recv(1024)
                if not chunk:
                    break
                response_bytes += chunk
                if b"\n" in response_bytes:
                    break
                    
        response_line = response_bytes.split(b"\n")[0].decode('utf-8')
        response_dict = json.loads(response_line)
        
        if response_dict.get("status") != "success":
            raise RuntimeError(f"Daemon rejected request: {response_dict.get('reason')}")
            
        return response_dict

    def close(self):
        if self.conn:
            self.conn.close()
        if sys.platform == "win32" and hasattr(self, 'handle'):
            import win32file
            win32file.CloseHandle(self.handle)

# ==========================================
# Runtime Demo
# ==========================================
if __name__ == "__main__":
    try:
        client = KeychainAuthClient()
        
        # 1. Batch Write Environment Keys
        print("Writing environment keys to keychain...")
        write_req = {
            "type": "REQUEST",
            "action": "write",
            "service": "AgentSecrets",
            "targets": [
                "proj_123:development:DATABASE_URL",
                "proj_123:staging:DATABASE_URL",
                "proj_123:production:DATABASE_URL"
            ],
            "values": [
                "postgres://dev-db",
                "postgres://staging-db",
                "postgres://prod-db"
            ]
        }
        client.send_request(write_req)
        print("Write succeeded!")

        # 2. Single-Roundtrip Prefix Read
        print("\nReading staging configuration via Prefix Read...")
        read_req = {
            "type": "REQUEST",
            "action": "read",
            "service": "AgentSecrets",
            "match": "prefix",
            "targets": ["proj_123:staging:"]
        }
        response = client.send_request(read_req)
        
        for item in response.get("results", []):
            print(f" -> Secret Key: {item['target']} = {item['value']}")
            
    except Exception as e:
        print(f"Error: {e}")

Key sections explained

1. Platform detection

   The _connect method checks sys.platform to branch between Unix and Windows connection logic:

   if sys.platform == "win32":
       # Windows Named Pipe path
   else:
       # macOS / Linux Unix Domain Socket path
   

   This single check drives the entire transport difference. The rest of the class is symmetric — send_request applies the same sys.platform == "win32" check when choosing between win32file calls and raw socket I/O.

2. Unix socket connection with FD_CLOEXEC enforcement

   On macOS and Linux, the client creates a standard AF_UNIX socket then immediately sets the FD_CLOEXEC flag using fcntl. Python’s socket module does not set this flag by default.

   self.conn = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
   
   import fcntl
   flags = fcntl.fcntl(self.conn.fileno(), fcntl.F_GETFD)
   fcntl.fcntl(self.conn.fileno(), fcntl.F_SETFD, flags | fcntl.FD_CLOEXEC)
   
   self.conn.connect(socket_path)
   

   The existing flags are read with F_GETFD first so that only FD_CLOEXEC is added, leaving any other descriptor flags intact. The flag is set before connect() to avoid any window where the descriptor could be inherited.

   The client selects the socket path based on the platform: on macOS it uses ~/Library/Application Support/keychain-auth/agent.sock; on Linux it uses $XDG_RUNTIME_DIR/keychain-auth/agent.sock (falling back to ~/.cache/keychain-auth/agent.sock):

   import platform
   if platform.system() == "Darwin":
       socket_path = os.path.expanduser("~/Library/Application Support/keychain-auth/agent.sock")
   else:
       runtime_dir = os.environ.get("XDG_RUNTIME_DIR", os.path.expanduser("~/.cache"))
       socket_path = os.path.join(runtime_dir, "keychain-auth", "agent.sock")
   
3. Windows Named Pipe via pywin32

   On Windows, the client opens the Named Pipe using win32file.CreateFile from the pywin32 library. The pipe path is always \\.\pipe\keychain-auth.

   import win32file
   import win32pipe
   self.handle = win32file.CreateFile(
       self.pipe_path,
       win32file.GENERIC_READ | win32file.GENERIC_WRITE,
       0, None, win32file.OPEN_EXISTING, 0, None
   )
   

   The ImportError catch provides a clear message when pywin32 is absent:

   except ImportError:
       raise RuntimeError("Please install 'pywin32' package for named pipe connections on Windows.")
   

   On Windows you must install the pywin32 package before using this client: pip install pywin32. The win32file and win32pipe modules it provides are the only supported way to interact with Windows Named Pipes from Python.

4. Sending JSON requests (newline-terminated)

   Every request is a JSON object serialized to a single line and terminated with \n. The daemon’s parser requires this terminator to detect the end of the message.

   def send_request(self, request_dict):
       payload = json.dumps(request_dict) + "\n"
   

   On Unix, the full payload is sent with sendall to guarantee all bytes are written:

   self.conn.sendall(payload.encode('utf-8'))
   

   On Windows, it is written to the pipe handle with win32file.WriteFile:

   win32file.WriteFile(self.handle, payload.encode('utf-8'))
   

   Both paths encode the string as UTF-8 bytes before transmission.

5. Parsing the response and error handling

   The client reads chunks in a loop until it receives a newline, then splits on \n and decodes only the first line:

   response_bytes = b""
   while True:
       chunk = self.conn.recv(1024)
       if not chunk:
           break
       response_bytes += chunk
       if b"\n" in response_bytes:
           break
   
   response_line = response_bytes.split(b"\n")[0].decode('utf-8')
   response_dict = json.loads(response_line)
   

   After parsing, the status field is checked. Any non-"success" status raises a RuntimeError that includes the daemon’s reason code, giving callers a machine-readable error string:

   if response_dict.get("status") != "success":
       raise RuntimeError(f"Daemon rejected request: {response_dict.get('reason')}")
   

   Callers should catch RuntimeError and inspect the message for reason codes such as unregistered_binary_pending_approval or service_not_allowed to produce actionable user-facing messages.