From 2b2bcd894e0ddbf8030ef2c9513380ecc5543170 Mon Sep 17 00:00:00 2001
From: Florian Best <best@univention.de>
Date: Wed, 3 Nov 2021 14:59:03 +0100
Subject: [PATCH] security: check if shared memory belongs to current user and
 is only read/writeable for them

Prevents
1. information disclosure
2. unpickling of untrusted pickle files resulting in code execution
vulnerabilities

Execute as user `nobody`:
```
$ python3
>>> with open('/dev/shm/sm_foo', 'wb') as fd:
...  fd.write(b'\x80\x03csubprocess\ncall\nq\x00X\n\x00\x00\x00/bin/touchq\x01X\x0b\x00\x00\x00/tmp/hackedq\x02\x86q\x03\x85q\x04Rq\x05.')
...
66
$ ls -l '/dev/shm/sm_foo'
-rw-r--r-- 1 nobody nogroup 66 Okt 21 18:42 /dev/shm/sm_foo
```

Then execute a new process as any user (e.g. root):

```
$ python3
>>> import shared_memory_dict
>>> f = shared_memory_dict.SharedMemoryDict('foo', 500)
>>> f
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "/home/fbest/git/shared-memory-dict/shared_memory_dict/dict.py", line 115, in __repr__
    return repr(self._read_memory())
  File "/home/fbest/git/shared-memory-dict/shared_memory_dict/dict.py", line 169, in _read_memory
    db = {key: self._unmap_value(key, value) for key, value in db.items()}
AttributeError: 'int' object has no attribute 'items'

$ ls -l /tmp/hacked
-rw-r--r-- 1 root root 0 Okt 21 18:45 /tmp/hacked
```

The command /bin/touch /tmp/hacked has been executed as root.

Fixes #33
---
 shared_memory_dict/dict.py | 19 +++++++++++++++++++
 1 file changed, 19 insertions(+)

diff --git a/shared_memory_dict/dict.py b/shared_memory_dict/dict.py
index ba01f95..1eb6202 100644
--- a/shared_memory_dict/dict.py
+++ b/shared_memory_dict/dict.py
@@ -1,4 +1,5 @@
 import logging
+import os
 import sys
 import warnings
 from contextlib import contextmanager
@@ -159,11 +160,29 @@ def _get_or_create_memory_block(
         try:
             return SharedMemory(name=name)
         except FileNotFoundError:
+            self.check_security(name)
             shm = SharedMemory(name=name, create=True, size=size)
             data = self._serializer.dumps({})
             shm.buf[: len(data)] = data
             return shm
 
+    def check_security(self, name: str) -> None:
+        """Check if shared memory belongs to the current user and is only read+writeable for us"""
+        if os.name == 'nt':
+            return
+
+        if '/' in name:
+            raise TypeError('Name must not contain "/".')
+
+        shm_file = os.path.join('/dev/shm', name)
+        stat = os.stat(shm_file)
+        if (
+            stat.st_uid != os.getuid()
+            or stat.st_gid != os.getgid()
+            or stat.st_mode != 0o100600
+        ):
+            os.unlink(shm_file)
+
     def _save_memory(self, db: Dict[str, Any]) -> None:
         data = self._serializer.dumps(db)
         try: