memory backed ephermal session store for easy concurrent runs

Change-Id: I05bfd6ff471ccf1c8c2b5f2b748b9d4125ddd4f7

osmo-smdpp.py

@@ -405,7 +405,7 @@ class SmDppHttpServer:
                     continue
         return False
 
-    def __init__(self, server_hostname: str, ci_certs_path: str, common_cert_path: str, use_brainpool: bool = False):
+    def __init__(self, server_hostname: str, ci_certs_path: str, common_cert_path: str, use_brainpool: bool = False, in_memory: bool = False):
         self.server_hostname = server_hostname
         self.upp_dir = os.path.realpath(os.path.join(DATA_DIR, 'upp'))
         self.ci_certs = self.load_certs_from_path(ci_certs_path)
@@ -426,9 +426,15 @@ class SmDppHttpServer:
         else:
             self.dp_pb.cert_from_der_file(os.path.join(cert_dir, 'DPpb', 'CERT_S_SM_DPpb_ECDSA_NIST.der'))
             self.dp_pb.privkey_from_pem_file(os.path.join(cert_dir, 'DPpb', 'SK_S_SM_DPpb_ECDSA_NIST.pem'))
+        if in_memory:
+            self.rss = rsp.RspSessionStore(in_memory=True)
+            logger.info("Using in-memory session storage")
+        else:
             # Use different session database files for BRP and NIST to avoid file locking during concurrent runs
             session_db_suffix = "BRP" if use_brainpool else "NIST"
-        self.rss = rsp.RspSessionStore(os.path.join(DATA_DIR, f"sm-dp-sessions-{session_db_suffix}"))
+            db_path = os.path.join(DATA_DIR, f"sm-dp-sessions-{session_db_suffix}")
+            self.rss = rsp.RspSessionStore(filename=db_path, in_memory=False)
+            logger.info(f"Using file-based session storage: {db_path}")
 
     @app.handle_errors(ApiError)
     def handle_apierror(self, request: IRequest, failure):
@@ -861,6 +867,8 @@ def main(argv):
     parser.add_argument("-v", "--verbose", help="dump more raw info", action='store_true', default=False)
     parser.add_argument("-b", "--brainpool", help="Use Brainpool curves instead of NIST",
                         action='store_true', default=False)
+    parser.add_argument("-m", "--in-memory", help="Use ephermal in-memory session storage (for concurrent runs)",
+                        action='store_true', default=False)
     args = parser.parse_args()
 
     logging.basicConfig(level=logging.DEBUG if args.verbose else logging.WARNING)

pySim/esim/rsp.py

@@ -94,9 +94,57 @@ class RspSessionState:
         self.__dict__.update(state)
 
 
-class RspSessionStore(shelve.DbfilenameShelf):
+class RspSessionStore:
-    """A derived class as wrapper around the database-backed non-volatile storage 'shelve', in case we might
+    """A wrapper around the database-backed storage 'shelve' for storing RspSessionState objects.
-    need to extend it in the future. We use it to store RspSessionState objects indexed by transactionId."""
+    Can be configured to use either file-based storage or in-memory storage.
+    We use it to store RspSessionState objects indexed by transactionId."""
+
+    def __init__(self, filename: Optional[str] = None, in_memory: bool = False):
+        self._in_memory = in_memory
+
+        if in_memory:
+            self._shelf = shelve.Shelf(dict())
+        else:
+            if filename is None:
+                raise ValueError("filename is required for file-based session store")
+            self._shelf = shelve.open(filename)
+
+    # dunder magic
+    def __getitem__(self, key):
+        return self._shelf[key]
+
+    def __setitem__(self, key, value):
+        self._shelf[key] = value
+
+    def __delitem__(self, key):
+        del self._shelf[key]
+
+    def __contains__(self, key):
+        return key in self._shelf
+
+    def __iter__(self):
+        return iter(self._shelf)
+
+    def __len__(self):
+        return len(self._shelf)
+
+    # everything else
+    def __getattr__(self, name):
+        """Delegate attribute access to the underlying shelf object."""
+        return getattr(self._shelf, name)
+
+    def close(self):
+        """Close the session store."""
+        if hasattr(self._shelf, 'close'):
+            self._shelf.close()
+        if self._in_memory:
+            # For in-memory store, clear the reference
+            self._shelf = None
+
+    def sync(self):
+        """Synchronize the cache with the underlying storage."""
+        if hasattr(self._shelf, 'sync'):
+            self._shelf.sync()
 
 def extract_euiccSigned1(authenticateServerResponse: bytes) -> bytes:
     """Extract the raw, DER-encoded binary euiccSigned1 field from the given AuthenticateServerResponse. This