From 1e4507c3d65248d1653db35b02f984af240ad492 Mon Sep 17 00:00:00 2001
From: Christian Kellner <christian@kellner.me>
Date: Wed, 27 Apr 2022 21:32:18 +0200
Subject: [PATCH] util/ostree: new class to store subordinate ids

Add a new class `SubIdsDB` as a database of subordinate Ids, like the
ones in `/etc/subuid` and `/etc/subgid`. Methods to read and write
data from these two files are provided.
Add corresponding unit tests.
---
 osbuild/util/ostree.py       | 50 +++++++++++++++++++++++++++++++++++
 test/mod/test_util_ostree.py | 51 ++++++++++++++++++++++++++++++++++++
 2 files changed, 101 insertions(+)

diff --git a/osbuild/util/ostree.py b/osbuild/util/ostree.py
index ad409470..84076a79 100644
--- a/osbuild/util/ostree.py
+++ b/osbuild/util/ostree.py
@@ -1,7 +1,9 @@
+import collections
 import contextlib
 import json
 import os
 import subprocess
+import sys
 import tempfile
 import typing
 
@@ -186,3 +188,51 @@ class PasswdLike:
     def _passwd_lines_to_dict(lines):
         """Take a list of passwd lines and produce a "name": "line" dictionary"""
         return {line.split(':')[0]: line for line in lines}
+
+
+class SubIdsDB:
+    """Represention of subordinate Ids database
+
+    Class to represent a mapping of a user name to subordinate ids,
+    like `/etc/subgid` and `/etc/subuid`.
+    """
+
+    def __init__(self) -> None:
+        self.db = collections.OrderedDict()
+
+    def read(self, fp) -> int:
+        idx = 0
+        for idx, line in enumerate(fp.readlines()):
+            line = line.strip()
+            if not line or line.startswith("#"):
+                continue
+            comps = line.split(":")
+            if len(comps) != 3:
+                print(f"WARNING: invalid line `{line}`", file=sys.stderr)
+                continue
+            name, uid, count = comps
+            self.db[name] = (uid, count)
+        return idx
+
+    def dumps(self) -> str:
+        """Dump the database to a string"""
+        data = "\n".join([
+            f"{name}:{uid}:{count}\n"
+            for name, (uid, count) in self.db.items()
+        ])
+
+        return data
+
+    def read_from(self, path: PathLike) -> int:
+        """Read a file and add the entries to the database"""
+        with open(path, "r", encoding="utf-8") as f:
+            return self.read(f)
+
+    def write_to(self, path: PathLike) -> None:
+        """Write the database to a file"""
+        data = self.dumps()
+        with open(path, "w", encoding="utf-8") as f:
+            f.write(data)
+
+    def __bool__(self) -> bool:
+        return bool(self.db)
diff --git a/test/mod/test_util_ostree.py b/test/mod/test_util_ostree.py
index 96584440..d6d4a463 100644
--- a/test/mod/test_util_ostree.py
+++ b/test/mod/test_util_ostree.py
@@ -2,6 +2,7 @@
 # Tests for the 'osbuild.util.ostree' module.
 #
 
+import io
 import json
 import os
 import subprocess
@@ -147,3 +148,53 @@ class TestPasswdLike(unittest.TestCase):
 
             with open(os.path.join(tmpdir, "result"), "r") as f:
                 self.assertEqual(sorted(f.readlines()), sorted(result_file_lines))
+
+    #pylint: disable=no-self-use
+    def test_subids_cfg(self):
+        with tempfile.TemporaryDirectory() as tmpdir:
+
+            first = [
+                "gicmo:100000:65536",
+                "achilles:100000:65536",
+                "ondrej:100000:65536"
+            ]
+
+            txt = io.StringIO("\n".join(first))
+
+            subids = ostree.SubIdsDB()
+            subids.read(txt)
+
+            assert len(subids.db) == 3
+
+            for name in ("gicmo", "achilles", "ondrej"):
+                assert name in subids.db
+                uid, count = subids.db[name]
+                assert uid == "100000"
+                assert count == "65536"
+
+            second = [
+                "gicmo:200000:1000",
+                "tom:200000:1000",
+                "lars:200000:1000"
+            ]
+
+            txt = io.StringIO("\n".join(second))
+            subids.read(txt)
+
+            assert len(subids.db) == 5
+
+            for name in ("gicmo", "achilles", "tom", "lars"):
+                assert name in subids.db
+
+            for name in ("gicmo", "tom", "lars"):
+                uid, count = subids.db[name]
+                assert uid == "200000"
+                assert count == "1000"
+
+            file = os.path.join(tmpdir, "subuid")
+            subids.write_to(file)
+
+            check = ostree.SubIdsDB()
+            check.read_from(file)
+
+            assert subids.db == check.db