openpgp: Convert verify/sign API to use string I/O

Use string I/O for consistency with the API Manifests use where we
expect to work on cleartext anyway. Bytes are now used only for loading
keys which can be in binary format.

3 files changed, 26 insertions, 30 deletions

diff --git a/gemato/manifest.py b/gemato/manifest.py
index 3743d5c..babd3b3 100644
--- a/gemato/manifest.py
+++ b/gemato/manifest.py
@@ -383,7 +383,7 @@ class ManifestFile(object):
                     "Manifest terminated early, inside signature")
 
         if verify_openpgp and state == ManifestState.POST_SIGNED_DATA:
-            with io.BytesIO(openpgp_data.encode('utf8')) as f:
+            with io.StringIO(openpgp_data) as f:
                 gemato.openpgp.verify_file(f, env=openpgp_env)
             self.openpgp_signed = True
 
diff --git a/gemato/openpgp.py b/gemato/openpgp.py
index 7f4fc98..123f3fc 100644
--- a/gemato/openpgp.py
+++ b/gemato/openpgp.py
@@ -94,7 +94,7 @@ class OpenPGPEnvironment(object):
 def verify_file(f, env=None):
     """
     Perform an OpenPGP verification of Manifest data in open file @f.
-    The file should be open in binary mode and set at the beginning
+    The file should be open in text mode and set at the beginning
     (or start of signed part). Raises an exception if the verification
     fails.
 
@@ -105,7 +105,7 @@ def verify_file(f, env=None):
 
     exitst, out, err = _spawn_gpg(['--verify'],
             env.home if env is not None else None,
-            f.read())
+            f.read().encode('utf8'))
     if exitst != 0:
         raise gemato.exceptions.OpenPGPVerificationFailure(err.decode('utf8'))
 
@@ -114,7 +114,7 @@ def clear_sign_file(f, outf, keyid=None, env=None):
     """
     Create an OpenPGP cleartext signed message containing the data
     from open file @f, and writing it into open file @outf.
-    Both files should be open in binary mode and set at the appropriate
+    Both files should be open in text mode and set at the appropriate
     position. Raises an exception if signing fails.
 
     Pass @keyid to specify the key to use. If not specified,
@@ -127,8 +127,8 @@ def clear_sign_file(f, outf, keyid=None, env=None):
         args += ['--local-user', keyid]
     exitst, out, err = _spawn_gpg(['--clearsign'] + args,
             env.home if env is not None else None,
-            f.read())
+            f.read().encode('utf8'))
     if exitst != 0:
         raise gemato.exceptions.OpenPGPSigningFailure(err.decode('utf8'))
 
-    outf.write(out)
+    outf.write(out.decode('utf8'))
diff --git a/tests/test_openpgp.py b/tests/test_openpgp.py
index 0ee295d..7a48ea4 100644
--- a/tests/test_openpgp.py
+++ b/tests/test_openpgp.py
@@ -16,7 +16,7 @@ import gemato.openpgp
 import gemato.recursiveloader
 
 
-PUBLIC_KEY = u'''
+PUBLIC_KEY = b'''
 -----BEGIN PGP PUBLIC KEY BLOCK-----
 
 mQENBFnwXJMBCACgaTVz+d10TGL9zR920sb0GBFsitAJ5ZFzO4E0cg3SHhwI+reM
@@ -36,7 +36,7 @@ jCvJNJ7pU8YnJSRTQDH0PZEupAdzDU/AhGSrBz5+Jr7N0pQIxq4duE/Q
 -----END PGP PUBLIC KEY BLOCK-----
 '''
 
-PRIVATE_KEY = u'''
+PRIVATE_KEY = b'''
 -----BEGIN PGP PRIVATE KEY BLOCK-----
 
 lQOYBFnwXJMBCACgaTVz+d10TGL9zR920sb0GBFsitAJ5ZFzO4E0cg3SHhwI+reM
@@ -73,7 +73,7 @@ DU/AhGSrBz5+Jr7N0pQIxq4duE/Q
 
 PRIVATE_KEY_ID = b'0x136880E72A7B1384'
 
-MALFORMED_PUBLIC_KEY = u'''
+MALFORMED_PUBLIC_KEY = b'''
 -----BEGIN PGP PUBLIC KEY BLOCK-----
 
 mQENBFnwXJMBCACgaTVz+d10TGL9zR920sb0GBFsitAJ5ZFzO4E0cg3SHhwI+reM
@@ -278,8 +278,7 @@ class OpenPGPCorrectKeyTest(unittest.TestCase):
     def setUp(self):
         self.env = gemato.openpgp.OpenPGPEnvironment()
         try:
-            self.env.import_key(
-                    io.BytesIO(PUBLIC_KEY.encode('utf8')))
+            self.env.import_key(io.BytesIO(PUBLIC_KEY))
         except gemato.exceptions.OpenPGPNoImplementation as e:
             raise unittest.SkipTest(str(e))
         except RuntimeError:
@@ -289,15 +288,15 @@ class OpenPGPCorrectKeyTest(unittest.TestCase):
         self.env.close()
 
     def test_verify_manifest(self):
-        with io.BytesIO(SIGNED_MANIFEST.encode('utf8')) as f:
+        with io.StringIO(SIGNED_MANIFEST) as f:
             self.env.verify_file(f)
 
     def test_verify_dash_escaped_manifest(self):
-        with io.BytesIO(DASH_ESCAPED_SIGNED_MANIFEST.encode('utf8')) as f:
+        with io.StringIO(DASH_ESCAPED_SIGNED_MANIFEST) as f:
             self.env.verify_file(f)
 
     def test_verify_modified_manifest(self):
-        with io.BytesIO(MODIFIED_SIGNED_MANIFEST.encode('utf8')) as f:
+        with io.StringIO(MODIFIED_SIGNED_MANIFEST) as f:
             self.assertRaises(gemato.exceptions.OpenPGPVerificationFailure,
                     self.env.verify_file, f)
 
@@ -355,7 +354,7 @@ class OpenPGPCorrectKeyTest(unittest.TestCase):
     def test_cli(self):
         d = tempfile.mkdtemp()
         try:
-            with io.open(os.path.join(d, '.key.asc'), 'w') as f:
+            with io.open(os.path.join(d, '.key.asc'), 'wb') as f:
                 f.write(PUBLIC_KEY)
             with io.open(os.path.join(d, 'Manifest'), 'w') as f:
                 f.write(SIGNED_MANIFEST)
@@ -389,7 +388,7 @@ class OpenPGPNoKeyTest(unittest.TestCase):
         self.env.close()
 
     def test_verify_manifest(self):
-        with io.BytesIO(SIGNED_MANIFEST.encode('utf8')) as f:
+        with io.StringIO(SIGNED_MANIFEST) as f:
             try:
                 self.assertRaises(gemato.exceptions.OpenPGPVerificationFailure,
                         self.env.verify_file, f)
@@ -477,8 +476,7 @@ class OpenPGPContextManagerTest(unittest.TestCase):
     def test_import_key(self):
         with gemato.openpgp.OpenPGPEnvironment() as env:
             try:
-                env.import_key(
-                        io.BytesIO(PUBLIC_KEY.encode('utf8')))
+                env.import_key(io.BytesIO(PUBLIC_KEY))
             except gemato.exceptions.OpenPGPNoImplementation as e:
                 raise unittest.SkipTest(str(e))
 
@@ -487,7 +485,7 @@ class OpenPGPContextManagerTest(unittest.TestCase):
             try:
                 self.assertRaises(RuntimeError,
                         env.import_key,
-                        io.BytesIO(MALFORMED_PUBLIC_KEY.encode('utf8')))
+                        io.BytesIO(MALFORMED_PUBLIC_KEY))
             except gemato.exceptions.OpenPGPNoImplementation as e:
                 raise unittest.SkipTest(str(e))
 
@@ -501,12 +499,11 @@ class OpenPGPContextManagerTest(unittest.TestCase):
                 raise unittest.SkipTest(str(e))
 
     def test_verify_manifest(self):
-        with io.BytesIO(SIGNED_MANIFEST.encode('utf8')) as f:
+        with io.StringIO(SIGNED_MANIFEST) as f:
             with gemato.openpgp.OpenPGPEnvironment() as env:
                 try:
                     try:
-                        env.import_key(
-                                io.BytesIO(PUBLIC_KEY.encode('utf8')))
+                        env.import_key(io.BytesIO(PUBLIC_KEY))
                     except RuntimeError:
                         raise unittest.SkipTest('Unable to import OpenPGP key')
 
@@ -530,13 +527,12 @@ class OpenPGPPrivateKeyTest(unittest.TestCase):
     Tests performed with the private key available.
     """
 
-    TEST_STRING = b'The quick brown fox jumps over the lazy dog'
+    TEST_STRING = u'The quick brown fox jumps over the lazy dog'
 
     def setUp(self):
         self.env = gemato.openpgp.OpenPGPEnvironment()
         try:
-            self.env.import_key(
-                    io.BytesIO(PRIVATE_KEY.encode('utf8')))
+            self.env.import_key(io.BytesIO(PRIVATE_KEY))
         except gemato.exceptions.OpenPGPNoImplementation as e:
             raise unittest.SkipTest(str(e))
         except RuntimeError:
@@ -546,19 +542,19 @@ class OpenPGPPrivateKeyTest(unittest.TestCase):
         self.env.close()
 
     def test_verify_manifest(self):
-        with io.BytesIO(SIGNED_MANIFEST.encode('utf8')) as f:
+        with io.StringIO(SIGNED_MANIFEST) as f:
             self.env.verify_file(f)
 
     def test_sign_data(self):
-        with io.BytesIO(self.TEST_STRING) as f:
-            with io.BytesIO() as wf:
+        with io.StringIO(self.TEST_STRING) as f:
+            with io.StringIO() as wf:
                 self.env.clear_sign_file(f, wf)
                 wf.seek(0)
                 self.env.verify_file(wf)
 
     def test_sign_data_keyid(self):
-        with io.BytesIO(self.TEST_STRING) as f:
-            with io.BytesIO() as wf:
+        with io.StringIO(self.TEST_STRING) as f:
+            with io.StringIO() as wf:
                 self.env.clear_sign_file(f, wf, keyid=PRIVATE_KEY_ID)
                 wf.seek(0)
                 self.env.verify_file(wf)