Various improvment of macOS backend

keyring/backends/macOS/__init__.py

@@ -5,6 +5,7 @@
 
 from ...backend import KeyringBackend
 from ...compat import properties
+from ...credentials import SimpleCredential
 from ...errors import KeyringError, KeyringLocked, PasswordDeleteError, PasswordSetError
 
 try:
@@ -54,11 +55,9 @@ def set_password(self, service, username, password):
 
     @warn_keychain
     def get_password(self, service, username):
-        if username is None:
-            username = ''
-
         try:
-            return api.find_generic_password(self.keychain, service, username)
+            _, password = api.find_generic_password(self.keychain, service, username)
+            return password
         except api.NotFound:
             pass
         except api.KeychainDenied as e:
@@ -83,3 +82,15 @@ def with_keychain(self, keychain):
             stacklevel=2,
         )
         return self.with_properties(keychain=keychain)
+
+    @warn_keychain
+    def get_credential(self, service, username):
+        try:
+            username, password = api.find_generic_password(self.keychain, service, username)
+            return SimpleCredential(username, password)
+        except api.NotFound:
+            pass
+        except api.KeychainDenied as e:
+            raise KeyringLocked(f"Can't get credential from keychain: {e}") from e
+        except api.Error as e:
+            raise KeyringError(f"Can't get credential from keychain: {e}") from e

keyring/backends/macOS/api.py

@@ -5,7 +5,9 @@
 import functools
 from ctypes import (
     byref,
+    c_bool,
     c_int32,
+    c_long,
     c_uint32,
     c_void_p,
 )
@@ -21,6 +23,7 @@ class error:
     plist_missing = -67030
     sec_interaction_not_allowed = -25308
 
+kCFStringEncodingUTF8 = 0x08000100
 
 _sec = ctypes.CDLL(find_library('Security'))
 _core = ctypes.CDLL(find_library('CoreServices'))
@@ -32,7 +35,7 @@ class error:
     c_void_p,
     c_void_p,
     c_void_p,
-    c_int32,
+    c_long,
     c_void_p,
     c_void_p,
 )
@@ -43,7 +46,7 @@ class error:
 
 CFNumberCreate = _found.CFNumberCreate
 CFNumberCreate.restype = c_void_p
-CFNumberCreate.argtypes = [c_void_p, c_uint32, ctypes.c_void_p]
+CFNumberCreate.argtypes = [c_void_p, c_long, ctypes.c_void_p]
 
 SecItemAdd = _sec.SecItemAdd
 SecItemAdd.restype = OS_status
@@ -62,9 +65,50 @@ class error:
 CFDataGetBytePtr.argtypes = (c_void_p,)
 
 CFDataGetLength = _found.CFDataGetLength
-CFDataGetLength.restype = c_int32
+CFDataGetLength.restype = c_long
 CFDataGetLength.argtypes = (c_void_p,)
 
+CFStringGetLength = _found.CFStringGetLength
+CFStringGetLength.restype = c_long
+CFStringGetLength.argtypes = (
+    c_void_p,
+)
+
+CFStringGetMaximumSizeForEncoding = _found.CFStringGetMaximumSizeForEncoding
+CFStringGetMaximumSizeForEncoding.restype = c_long
+CFStringGetMaximumSizeForEncoding.argtypes = (
+    c_long,
+    c_uint32
+)
+
+CFStringGetCString = _found.CFStringGetCString
+CFStringGetCString.restype = c_bool
+CFStringGetCString.argtypes = (
+    c_void_p,
+    c_void_p,
+    c_long,
+    c_uint32
+)
+
+CFDictionaryContainsKey = _found.CFDictionaryContainsKey
+CFDictionaryContainsKey.restype = c_bool
+CFDictionaryContainsKey.argtypes = (
+    c_void_p,
+    c_void_p,
+)
+
+CFDictionaryGetValue = _found.CFDictionaryGetValue
+CFDictionaryGetValue.restype = c_void_p
+CFDictionaryGetValue.argtypes = (
+    c_void_p,
+    c_void_p,
+)
+
+CFRelease = _found.CFRelease
+CFRelease.restype = None
+CFRelease.argtypes = (
+    c_void_p,
+)
 
 def k_(s):
     return c_void_p.in_dll(_sec, s)
@@ -79,34 +123,54 @@ def create_cf(ob):
 @create_cf.register(bool)
 @create_cf.register(int)
 def _(val: bool | int):
+    kCFNumberSInt32Type = 3
     if val.bit_length() > 31:
         raise OverflowError(val)
-    int32 = 0x9
-    return CFNumberCreate(None, int32, ctypes.byref(c_int32(val)))
+    return CFNumberCreate(None, kCFNumberSInt32Type, ctypes.byref(c_int32(val)))
 
 
 @create_cf.register
 def _(s: str):
-    kCFStringEncodingUTF8 = 0x08000100
     return CFStringCreateWithCString(None, s.encode('utf8'), kCFStringEncodingUTF8)
 
 
 def create_query(**kwargs):
-    return CFDictionaryCreate(
-        None,
-        (c_void_p * len(kwargs))(*map(k_, kwargs.keys())),
-        (c_void_p * len(kwargs))(*map(create_cf, kwargs.values())),
-        len(kwargs),
-        _found.kCFTypeDictionaryKeyCallBacks,
-        _found.kCFTypeDictionaryValueCallBacks,
-    )
-
+    values_cf = []
+    try:
+        values_cf.extend(create_cf(v) for v in kwargs.values())
+
+        cf_dict = CFDictionaryCreate(
+            None,
+            (c_void_p * len(kwargs))(*map(k_, kwargs.keys())),
+            (c_void_p * len(kwargs))(*values_cf),
+            len(kwargs),
+            _found.kCFTypeDictionaryKeyCallBacks,
+            _found.kCFTypeDictionaryValueCallBacks,
+        )
+
+        return cf_dict
+
+    finally:
+        # Free memory
+        for cf in values_cf:
+            # Note: some values here are borrowed CF constants (e.g. kSecClass)
+            # We don't own them and releasing them would normally be an over-release.
+            # Since they are implemented as immortal CF objects,
+            # CFRelease on them is a silent no-op.
+            CFRelease(cf)
 
 def cfstr_to_str(data):
     return ctypes.string_at(CFDataGetBytePtr(data), CFDataGetLength(data)).decode(
         'utf-8'
     )
 
+def cfstring_to_str(cf_string):
+    str_length = CFStringGetLength(cf_string)
+    str_max_size = CFStringGetMaximumSizeForEncoding(str_length, kCFStringEncodingUTF8) + 1
+    buffer = ctypes.create_string_buffer(str_max_size)
+    if CFStringGetCString(cf_string, buffer, str_max_size, kCFStringEncodingUTF8):
+        return ctypes.string_at(buffer).decode('utf-8')
+    return None
 
 class Error(Exception):
     @classmethod
@@ -138,47 +202,88 @@ class SecAuthFailure(Error):
     pass
 
 
-def find_generic_password(kc_name, service, username, not_found_ok=False):
-    q = create_query(
-        kSecClass=k_('kSecClassGenericPassword'),
-        kSecMatchLimit=k_('kSecMatchLimitOne'),
-        kSecAttrService=service,
-        kSecAttrAccount=username,
-        kSecReturnData=True,
-    )
-
+def find_generic_password(kc_name, service, username):
+    q = None
     data = c_void_p()
-    status = SecItemCopyMatching(q, byref(data))
 
-    if status == error.item_not_found and not_found_ok:
-        return
+    try:
+        query = dict(
+            kSecClass=k_('kSecClassGenericPassword'),
+            kSecMatchLimit=k_('kSecMatchLimitOne'),
+            kSecAttrService=service,
+            kSecReturnAttributes=True,
+            kSecReturnData=True,
+        )
+
+        # Use the username in the query, if provided
+        if bool(username):
+            query = dict(
+                query,
+                kSecAttrAccount=username,
+            )
+
+        q = create_query(**query)
+
+        status = SecItemCopyMatching(q, byref(data))
+
+        Error.raise_for_status(status)
+
+        # Extract username and password from the query
+        ret_username = None
+        password = None
+        if CFDictionaryContainsKey(data, k_('kSecAttrAccount')):
+            ret = CFDictionaryGetValue(data, k_('kSecAttrAccount'))
+            ret_username = cfstring_to_str(ret)
 
-    Error.raise_for_status(status)
+        if CFDictionaryContainsKey(data, k_('kSecValueData')):
+            ret = CFDictionaryGetValue(data, k_('kSecValueData'))
+            password = cfstr_to_str(ret)
 
-    return cfstr_to_str(data)
+        return ret_username, password
+    finally:
+        # Free memory
+        if q:
+            CFRelease(q)
+        if data.value:
+            CFRelease(data)
 
 
 def set_generic_password(name, service, username, password):
     with contextlib.suppress(NotFound):
         delete_generic_password(name, service, username)
 
-    q = create_query(
-        kSecClass=k_('kSecClassGenericPassword'),
-        kSecAttrService=service,
-        kSecAttrAccount=username,
-        kSecValueData=password,
-    )
+    q = None
 
-    status = SecItemAdd(q, None)
-    Error.raise_for_status(status)
+    try:
+        q = create_query(
+            kSecClass=k_('kSecClassGenericPassword'),
+            kSecAttrService=service,
+            kSecAttrAccount=username,
+            kSecValueData=password,
+        )
 
+        status = SecItemAdd(q, None)
+        Error.raise_for_status(status)
 
-def delete_generic_password(name, service, username):
-    q = create_query(
-        kSecClass=k_('kSecClassGenericPassword'),
-        kSecAttrService=service,
-        kSecAttrAccount=username,
-    )
+    finally:
+        # Free memory
+        if q:
+            CFRelease(q)
 
-    status = SecItemDelete(q)
-    Error.raise_for_status(status)
+def delete_generic_password(name, service, username):
+    q = None
+
+    try:
+        q = create_query(
+            kSecClass=k_('kSecClassGenericPassword'),
+            kSecAttrService=service,
+            kSecAttrAccount=username,
+        )
+
+        status = SecItemDelete(q)
+        Error.raise_for_status(status)
+
+    finally:
+        # Free memory
+        if q:
+            CFRelease(q)

keyring/testing/backend.py

@@ -151,7 +151,8 @@ def test_credential(self):
         self.set_password('service1', 'user2', 'password2')
 
         cred = keyring.get_credential('service1', None)
-        assert cred is None or (cred.username, cred.password) in (
+        assert cred is not None
+        assert (cred.username, cred.password) in (
             ('user1', 'password1'),
             ('user2', 'password2'),
         )