Добавлен объект для проверки GPG подписи

pym/calculate/lib/utils/gpg.py

@@ -0,0 +1,140 @@
+# -*- coding: utf-8 -*-
+
+# Copyright 2018 Mir Calculate. http://www.calculate-linux.org
+#
+#  Licensed under the Apache License, Version 2.0 (the "License");
+#  you may not use this file except in compliance with the License.
+#  You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#  See the License for the specific language governing permissions and
+#  limitations under the License.
+
+from calculate.lib.utils.files import (process, getProgPath, makeDirectory,
+    removeDir)
+import os
+
+class GPGError(Exception):
+    pass
+
+class GPGKeyExpired(GPGError):
+    pass
+
+class GPGKeyRevoced(GPGError):
+    pass
+
+class GPGNoKey(GPGError):
+    pass
+
+class GPGImportError(GPGError):
+    pass
+
+class Pipe(object):
+    def __init__(self):
+        self.outfd, self.infd = os.pipe()
+
+    def write(self, data):
+        os.write(self.infd, data)
+
+    def get_filename(self):
+        return "/proc/{}/fd/{}".format(os.getpid(), self.outfd)
+
+    def get_fd(self):
+        return self.outfd
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, exc_type, exc_value, exc_cb):
+        self.close()
+
+    def closein(self):
+        if self.infd is not None:
+            os.close(self.infd)
+            self.infd = None
+
+    def close(self):
+        self.closein()
+        if self.outfd is not None:
+            os.close(self.outfd)
+            self.outfd = None
+
+
+class GPG(object):
+    GNUPREFIX = "[GNUPG:]"
+    GOOD_PREFIX = "%s GOODSIG" % GNUPREFIX
+    EXPIRED_PREFIX = "%s EXPKEYSIG" % GNUPREFIX
+    REVOCED_PREFIX = "%s REVKEYSIG" % GNUPREFIX
+    SIGDATA_PREFIX = "%s VALIDSIG" % GNUPREFIX
+    
+    def __init__(self, homedir, timeout=20):
+        self.homedir = homedir
+        self.timeout = timeout
+        makeDirectory(self.homedir)
+
+    @property
+    def gpgcmd(self):
+        cmd = getProgPath("/usr/bin/gpg")
+        if not cmd:
+            raise GPGError(_("%s command not found"))
+        return cmd
+
+    def _spawn_gpg(self, *options):
+        return process(self.gpgcmd, "--status-fd", "1",
+                       "--batch", "--no-autostart",
+                       "--homedir", self.homedir, *options,
+                       timeout=self.timeout)
+
+    def recv_keys(self, keyid):
+        return _spawn_gpg("--recv-keys", keyid).success()
+
+    def import_key(self, keyfile):
+        p = self._spawn_gpg("--import")
+        p.write(keyfile.read())
+        if not p.success():
+            raise GPGImportError(p.readerr())
+
+    def verify(self, data, sign):
+        with Pipe() as datapipe, Pipe() as signpipe:
+            p = self._spawn_gpg("--verify",
+                                signpipe.get_filename(), datapipe.get_filename())
+            signpipe.write(sign.read())
+            signpipe.closein()
+            datapipe.write(data.read())
+            datapipe.closein()
+
+            goodsig = False
+            sig_data_ok = False
+            for l in p:
+                if l.startswith(self.GOOD_PREFIX):
+                    goodsig = True
+                if l.startswith(self.EXPIRED_PREFIX):
+                    raise GPGKeyExpired(p.readerr())
+                if l.startswith(self.REVOCED_PREFIX):
+                    raise GPGKeyRevoced(p.readerr())
+                if l.startswith(self.SIGDATA_PREFIX):
+                    if len(l.split(' ')) >= 12:
+                        sig_data_ok = True
+            if not p.success() or not goodsig or not sig_data_ok:
+                if "No public key" in p.readerr():
+                    raise GPGNoKey(p.readerr())
+                else:
+                    raise GPGError(p.readerr())
+
+    def keys_list(self):
+        pass
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, exc_type, exc_value, exc_cb):
+        self.close()
+
+    def close(self):
+        if self.homedir:
+            removeDir(self.homedir)
+            self.homedir = None