app, test: encapsulate FIFO cache and add test

securedrop/crypto_util.py

@@ -55,6 +55,32 @@ def monkey_patch_delete_handle_status(self, key, value):
 gnupg._parsers.DeleteResult._handle_status = monkey_patch_delete_handle_status
 
 
+class FIFOCache():
+    """
+    We implemented this simple cache instead of using functools.lru_cache
+    (this uses a different cache replacement policy (FIFO), but either
+    FIFO or LRU works for our key fingerprint cache)
+    due to the inability to remove an item from its cache.
+
+    See: https://bugs.python.org/issue28178
+    """
+    def __init__(self, maxsize: int):
+        self.maxsize = maxsize
+        self.cache = collections.OrderedDict()  # type: collections.OrderedDict
+
+    def get(self, item):
+        if item in self.cache:
+            return self.cache[item]
+
+    def put(self, item, value):
+        self.cache[item] = value
+        if len(self.cache) > self.maxsize:
+            self.cache.popitem(last=False)
+
+    def delete(self, item):
+        del self.cache[item]
+
+
 class CryptoException(Exception):
     pass
 
@@ -73,8 +99,8 @@ class CryptoUtil:
     # to set an expiration date.
     DEFAULT_KEY_EXPIRATION_DATE = '0'
 
-    keycache = collections.OrderedDict()  # type: collections.OrderedDict
     keycache_limit = 1000
+    keycache = FIFOCache(keycache_limit)
 
     def __init__(self,
                  scrypt_params,
@@ -228,20 +254,19 @@ def delete_reply_keypair(self, source_filesystem_id):
         temp_gpg = gnupg.GPG(binary='gpg2', homedir=self.gpg_key_dir)
         # The subkeys keyword argument deletes both secret and public keys.
         temp_gpg.delete_keys(key, secret=True, subkeys=True)
-        del self.keycache[source_filesystem_id]
+        self.keycache.delete(source_filesystem_id)
 
     def getkey(self, name):
-        if name in self.keycache:
-            return self.keycache[name]
+        fingerprint = self.keycache.get(name)
+        if fingerprint:  # cache hit
+            return fingerprint
 
+        # cache miss
         for key in self.gpg.list_keys():
             for uid in key['uids']:
                 if name in uid:
-                    fingerprint = key['fingerprint']
-                    self.keycache[name] = fingerprint
-                    if len(self.keycache) > self.keycache_limit:
-                        self.keycache.popitem(last=False)
-                    return fingerprint
+                    self.keycache.put(name, key['fingerprint'])
+                    return key['fingerprint']
 
         return None
 

securedrop/tests/test_crypto_util.py

@@ -11,7 +11,7 @@
 import crypto_util
 import models
 
-from crypto_util import CryptoUtil, CryptoException
+from crypto_util import CryptoUtil, CryptoException, FIFOCache
 from db import db
 
 
@@ -343,3 +343,25 @@ def test_encrypt_then_decrypt_gives_same_result(
     decrypted_text = crypto.decrypt(secret, ciphertext)
 
     assert decrypted_text == message
+
+
+def test_fifo_cache():
+    cache = FIFOCache(3)
+
+    cache.put('item 1', 1)
+    cache.put('item 2', 2)
+    cache.put('item 3', 3)
+
+    assert cache.get('item 1') == 1
+    assert cache.get('item 2') == 2
+    assert cache.get('item 3') == 3
+
+    cache.put('item 4', 4)
+    # Maxsize is 3, so adding item 4 should kick out item 1
+    assert not cache.get('item 1')
+    assert cache.get('item 2') == 2
+    assert cache.get('item 3') == 3
+    assert cache.get('item 4') == 4
+
+    cache.delete('item 2')
+    assert not cache.get('item 2')