GpgFrontend/gpgme
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

python: Fix teardown of ephemeral contexts.

Browse Source 
* lang/python/tests/support.py (EphemeralContext): New function.
* lang/python/tests/t-quick-key-creation.py: Use the new function to
manage ephemeral contexts.
* lang/python/tests/t-quick-key-manipulation.py: Likewise.
* lang/python/tests/t-quick-subkey-creation.py: Likewise.
--

Previously, there was a problem with cleaning up ephemeral home
directories.  shutil.rmtree deleted the agents main socket, gpg-agent
detected that, and deleted the other sockets as well, racing
shutil.rmtree which did not cope will with that.

Fix this by asking the agent nicely to shut down.

Signed-off-by: Justus Winter <justus@g10code.com>
This commit is contained in:
Justus Winter 2017-02-17 12:18:56 +01:00
parent 9350168a1e
commit de8494b16b
No known key found for this signature in database
GPG Key ID: DD1A52F9DA8C9020
4 changed files with 232 additions and 242 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

24
lang/python/tests/support.py
View File

@ -18,9 +18,12 @@
from __future__ import absolute_import, print_function, unicode_literals from __future__ import absolute_import, print_function, unicode_literals
del absolute_import, print_function, unicode_literals del absolute_import, print_function, unicode_literals
import contextlib
import shutil
import sys import sys
import os import os
import tempfile import tempfile
import time
import gpg import gpg
# known keys # known keys
@ -85,5 +88,24 @@ else:
self.path = tempfile.mkdtemp() self.path = tempfile.mkdtemp()
return self.path return self.path
def __exit__(self, *args): def __exit__(self, *args):
import shutil
shutil.rmtree(self.path) shutil.rmtree(self.path)
@contextlib.contextmanager
def EphemeralContext():
with TemporaryDirectory() as tmp:
home = os.environ['GNUPGHOME']
shutil.copy(os.path.join(home, "gpg.conf"), tmp)
shutil.copy(os.path.join(home, "gpg-agent.conf"), tmp)
with gpg.Context(home_dir=tmp) as ctx:
yield ctx
# Ask the agent to quit.
agent_socket = os.path.join(tmp, "S.gpg-agent")
ctx.protocol = gpg.constants.protocol.ASSUAN
ctx.set_engine_info(ctx.protocol, file_name=agent_socket)
ctx.assuan_transact(["KILLAGENT"])
# Block until it is really gone.
while os.path.exists(agent_socket):
time.sleep(.01)

184
lang/python/tests/t-quick-key-creation.py
View File

@ -22,42 +22,33 @@ del absolute_import, print_function, unicode_literals
import gpg import gpg
import itertools import itertools
import os
import shutil
import time import time
import support import support
alpha = "Alpha <alpha@invalid.example.net>" alpha = "Alpha <alpha@invalid.example.net>"
def copy_configuration(destination): with support.EphemeralContext() as ctx:
home = os.environ['GNUPGHOME'] res = ctx.create_key(alpha)
shutil.copy(os.path.join(home, "gpg.conf"), destination)
shutil.copy(os.path.join(home, "gpg-agent.conf"), destination)
with support.TemporaryDirectory() as tmp: keys = list(ctx.keylist())
copy_configuration(tmp) assert len(keys) == 1, "Weird number of keys created"
with gpg.Context(home_dir=tmp) as ctx:
res = ctx.create_key(alpha)
keys = list(ctx.keylist()) key = keys[0]
assert len(keys) == 1, "Weird number of keys created" assert key.fpr == res.fpr
assert len(key.subkeys) == 2, "Expected one primary key and one subkey"
assert key.subkeys[0].expires > 0, "Expected primary key to expire"
key = keys[0] # Try to create a key with the same UID
assert key.fpr == res.fpr try:
assert len(key.subkeys) == 2, "Expected one primary key and one subkey" ctx.create_key(alpha)
assert key.subkeys[0].expires > 0, "Expected primary key to expire" assert False, "Expected an error but got none"
except gpg.errors.GpgError as e:
pass
# Try to create a key with the same UID # Try to create a key with the same UID, now with force!
try: res2 = ctx.create_key(alpha, force=True)
ctx.create_key(alpha) assert res.fpr != res2.fpr
assert False, "Expected an error but got none"
except gpg.errors.GpgError as e:
pass
# Try to create a key with the same UID, now with force!
res2 = ctx.create_key(alpha, force=True)
assert res.fpr != res2.fpr
# From here on, we use one context, and create unique UIDs # From here on, we use one context, and create unique UIDs
@ -67,85 +58,82 @@ def make_uid():
uid_counter += 1 uid_counter += 1
return "user{0}@invalid.example.org".format(uid_counter) return "user{0}@invalid.example.org".format(uid_counter)
with support.TemporaryDirectory() as tmp: with support.EphemeralContext() as ctx:
copy_configuration(tmp) # Check gpg.constants.create.NOEXPIRE...
with gpg.Context(home_dir=tmp) as ctx: res = ctx.create_key(make_uid(), expires=False)
key = ctx.get_key(res.fpr, secret=True)
assert key.fpr == res.fpr
assert len(key.subkeys) == 2, "Expected one primary key and one subkey"
assert key.subkeys[0].expires == 0, "Expected primary key not to expire"
# Check gpg.constants.create.NOEXPIRE... t = 2 * 24 * 60 * 60
res = ctx.create_key(make_uid(), expires=False) slack = 5 * 60
res = ctx.create_key(make_uid(), expires_in=t)
key = ctx.get_key(res.fpr, secret=True)
assert key.fpr == res.fpr
assert len(key.subkeys) == 2, "Expected one primary key and one subkey"
assert abs(time.time() + t - key.subkeys[0].expires) < slack, \
"Primary keys expiration time is off"
# Check capabilities
for sign, encrypt, certify, authenticate in itertools.product([False, True],
[False, True],
[False, True],
[False, True]):
# Filter some out
if not (sign or encrypt or certify or authenticate):
# This triggers the default capabilities tested before.
continue
if (sign or encrypt or authenticate) and not certify:
# The primary key always certifies.
continue
res = ctx.create_key(make_uid(), algorithm="rsa",
sign=sign, encrypt=encrypt, certify=certify,
authenticate=authenticate)
key = ctx.get_key(res.fpr, secret=True) key = ctx.get_key(res.fpr, secret=True)
assert key.fpr == res.fpr assert key.fpr == res.fpr
assert len(key.subkeys) == 2, "Expected one primary key and one subkey" assert len(key.subkeys) == 1, \
assert key.subkeys[0].expires == 0, "Expected primary key not to expire" "Expected no subkey for non-default capabilities"
t = 2 * 24 * 60 * 60 p = key.subkeys[0]
slack = 5 * 60 assert sign == p.can_sign
res = ctx.create_key(make_uid(), expires_in=t) assert encrypt == p.can_encrypt
key = ctx.get_key(res.fpr, secret=True) assert certify == p.can_certify
assert key.fpr == res.fpr assert authenticate == p.can_authenticate
assert len(key.subkeys) == 2, "Expected one primary key and one subkey"
assert abs(time.time() + t - key.subkeys[0].expires) < slack, \
"Primary keys expiration time is off"
# Check capabilities # Check algorithm
for sign, encrypt, certify, authenticate in itertools.product([False, True], res = ctx.create_key(make_uid(), algorithm="rsa")
[False, True], key = ctx.get_key(res.fpr, secret=True)
[False, True], assert key.fpr == res.fpr
[False, True]): for k in key.subkeys:
# Filter some out assert k.pubkey_algo == 1
if not (sign or encrypt or certify or authenticate):
# This triggers the default capabilities tested before.
continue
if (sign or encrypt or authenticate) and not certify:
# The primary key always certifies.
continue
res = ctx.create_key(make_uid(), algorithm="rsa", # Check algorithm with size
sign=sign, encrypt=encrypt, certify=certify, res = ctx.create_key(make_uid(), algorithm="rsa1024")
authenticate=authenticate) key = ctx.get_key(res.fpr, secret=True)
key = ctx.get_key(res.fpr, secret=True) assert key.fpr == res.fpr
assert key.fpr == res.fpr for k in key.subkeys:
assert len(key.subkeys) == 1, \ assert k.pubkey_algo == 1
"Expected no subkey for non-default capabilities" assert k.length == 1024
p = key.subkeys[0] # Check algorithm future-default
assert sign == p.can_sign ctx.create_key(make_uid(), algorithm="future-default")
assert encrypt == p.can_encrypt
assert certify == p.can_certify
assert authenticate == p.can_authenticate
# Check algorithm # Check passphrase protection
res = ctx.create_key(make_uid(), algorithm="rsa") recipient = make_uid()
key = ctx.get_key(res.fpr, secret=True) passphrase = "streng geheim"
assert key.fpr == res.fpr res = ctx.create_key(recipient, passphrase=passphrase)
for k in key.subkeys: ciphertext, _, _ = ctx.encrypt(b"hello there", recipients=[ctx.get_key(res.fpr)])
assert k.pubkey_algo == 1
# Check algorithm with size cb_called = False
res = ctx.create_key(make_uid(), algorithm="rsa1024") def cb(*args):
key = ctx.get_key(res.fpr, secret=True) global cb_called
assert key.fpr == res.fpr cb_called = True
for k in key.subkeys: return passphrase
assert k.pubkey_algo == 1 ctx.pinentry_mode = gpg.constants.PINENTRY_MODE_LOOPBACK
assert k.length == 1024 ctx.set_passphrase_cb(cb)
# Check algorithm future-default plaintext, _, _ = ctx.decrypt(ciphertext)
ctx.create_key(make_uid(), algorithm="future-default") assert plaintext == b"hello there"
assert cb_called
# Check passphrase protection
recipient = make_uid()
passphrase = "streng geheim"
res = ctx.create_key(recipient, passphrase=passphrase)
ciphertext, _, _ = ctx.encrypt(b"hello there", recipients=[ctx.get_key(res.fpr)])
cb_called = False
def cb(*args):
global cb_called
cb_called = True
return passphrase
ctx.pinentry_mode = gpg.constants.PINENTRY_MODE_LOOPBACK
ctx.set_passphrase_cb(cb)
plaintext, _, _ = ctx.decrypt(ciphertext)
assert plaintext == b"hello there"
assert cb_called

117
lang/python/tests/t-quick-key-manipulation.py
View File

@ -21,83 +21,72 @@ from __future__ import absolute_import, print_function, unicode_literals
del absolute_import, print_function, unicode_literals del absolute_import, print_function, unicode_literals
import gpg import gpg
import itertools
import os
import shutil
import time
import support import support
alpha = "Alpha <alpha@invalid.example.net>" alpha = "Alpha <alpha@invalid.example.net>"
bravo = "Bravo <bravo@invalid.example.net>" bravo = "Bravo <bravo@invalid.example.net>"
def copy_configuration(destination): with support.EphemeralContext() as ctx:
home = os.environ['GNUPGHOME'] res = ctx.create_key(alpha, certify=True)
shutil.copy(os.path.join(home, "gpg.conf"), destination) key = ctx.get_key(res.fpr)
shutil.copy(os.path.join(home, "gpg-agent.conf"), destination) assert len(key.subkeys) == 1, "Expected one primary key and no subkeys"
assert len(key.uids) == 1, "Expected exactly one UID"
with support.TemporaryDirectory() as tmp: def get_uid(uid):
copy_configuration(tmp)
with gpg.Context(home_dir=tmp) as ctx:
res = ctx.create_key(alpha, certify=True)
key = ctx.get_key(res.fpr) key = ctx.get_key(res.fpr)
assert len(key.subkeys) == 1, "Expected one primary key and no subkeys" for u in key.uids:
assert len(key.uids) == 1, "Expected exactly one UID" if u.uid == uid:
return u
return None
def get_uid(uid): # sanity check
key = ctx.get_key(res.fpr) uid = get_uid(alpha)
for u in key.uids: assert uid, "UID alpha not found"
if u.uid == uid: assert uid.revoked == 0
return u
return None
# sanity check # add bravo
uid = get_uid(alpha) ctx.key_add_uid(key, bravo)
assert uid, "UID alpha not found" uid = get_uid(bravo)
assert uid.revoked == 0 assert uid, "UID bravo not found"
assert uid.revoked == 0
# add bravo # revoke alpha
ctx.key_add_uid(key, bravo) ctx.key_revoke_uid(key, alpha)
uid = get_uid(bravo) uid = get_uid(alpha)
assert uid, "UID bravo not found" assert uid, "UID alpha not found"
assert uid.revoked == 0 assert uid.revoked == 1
uid = get_uid(bravo)
assert uid, "UID bravo not found"
assert uid.revoked == 0
# revoke alpha # try to revoke the last UID
try:
ctx.key_revoke_uid(key, alpha) ctx.key_revoke_uid(key, alpha)
uid = get_uid(alpha) # IMHO this should fail. issue2961.
assert uid, "UID alpha not found" # assert False, "Expected an error but got none"
assert uid.revoked == 1 except gpg.errors.GpgError:
uid = get_uid(bravo) pass
assert uid, "UID bravo not found"
assert uid.revoked == 0
# try to revoke the last UID # Everything should be the same
try: uid = get_uid(alpha)
ctx.key_revoke_uid(key, alpha) assert uid, "UID alpha not found"
# IMHO this should fail. issue2961. assert uid.revoked == 1
# assert False, "Expected an error but got none" uid = get_uid(bravo)
except gpg.errors.GpgError: assert uid, "UID bravo not found"
pass assert uid.revoked == 0
# Everything should be the same # try to revoke a non-existent UID
uid = get_uid(alpha) try:
assert uid, "UID alpha not found" ctx.key_revoke_uid(key, "i dont exist")
assert uid.revoked == 1 # IMHO this should fail. issue2963.
uid = get_uid(bravo) # assert False, "Expected an error but got none"
assert uid, "UID bravo not found" except gpg.errors.GpgError:
assert uid.revoked == 0 pass
# try to revoke a non-existent UID # try to add an pre-existent UID
try: try:
ctx.key_revoke_uid(key, "i dont exist") ctx.key_add_uid(key, bravo)
# IMHO this should fail. issue2963. assert False, "Expected an error but got none"
# assert False, "Expected an error but got none" except gpg.errors.GpgError:
except gpg.errors.GpgError: pass
pass
# try to add an pre-existent UID
try:
ctx.key_add_uid(key, bravo)
assert False, "Expected an error but got none"
except gpg.errors.GpgError:
pass

149
lang/python/tests/t-quick-subkey-creation.py
View File

@ -22,8 +22,6 @@ del absolute_import, print_function, unicode_literals
import gpg import gpg
import itertools import itertools
import os
import shutil
import time import time
import support import support
@ -31,91 +29,84 @@ import support
alpha = "Alpha <alpha@invalid.example.net>" alpha = "Alpha <alpha@invalid.example.net>"
bravo = "Bravo <bravo@invalid.example.net>" bravo = "Bravo <bravo@invalid.example.net>"
def copy_configuration(destination): with support.EphemeralContext() as ctx:
home = os.environ['GNUPGHOME'] res = ctx.create_key(alpha, certify=True)
shutil.copy(os.path.join(home, "gpg.conf"), destination) keys = list(ctx.keylist())
shutil.copy(os.path.join(home, "gpg-agent.conf"), destination) assert len(keys) == 1, "Weird number of keys created"
key = keys[0]
assert key.fpr == res.fpr
assert len(key.subkeys) == 1, "Expected one primary key and no subkeys"
with support.TemporaryDirectory() as tmp: def get_subkey(fpr):
copy_configuration(tmp) k = ctx.get_key(fpr)
with gpg.Context(home_dir=tmp) as ctx: for sk in k.subkeys:
res = ctx.create_key(alpha, certify=True) if sk.fpr == fpr:
keys = list(ctx.keylist()) return sk
assert len(keys) == 1, "Weird number of keys created" return None
key = keys[0]
assert key.fpr == res.fpr
assert len(key.subkeys) == 1, "Expected one primary key and no subkeys"
def get_subkey(fpr): # Check gpg.constants.create.NOEXPIRE...
k = ctx.get_key(fpr) res = ctx.create_subkey(key, expires=False)
for sk in k.subkeys: subkey = get_subkey(res.fpr)
if sk.fpr == fpr: assert subkey.expires == 0, "Expected subkey not to expire"
return sk assert subkey.can_encrypt, \
return None "Default subkey capabilities do not include encryption"
# Check gpg.constants.create.NOEXPIRE... t = 2 * 24 * 60 * 60
res = ctx.create_subkey(key, expires=False) slack = 5 * 60
res = ctx.create_subkey(key, expires_in=t)
subkey = get_subkey(res.fpr)
assert abs(time.time() + t - subkey.expires) < slack, \
"subkeys expiration time is off"
# Check capabilities
for sign, encrypt, authenticate in itertools.product([False, True],
[False, True],
[False, True]):
# Filter some out
if not (sign or encrypt or authenticate):
# This triggers the default capabilities tested before.
continue
res = ctx.create_subkey(key, sign=sign, encrypt=encrypt,
authenticate=authenticate)
subkey = get_subkey(res.fpr) subkey = get_subkey(res.fpr)
assert subkey.expires == 0, "Expected subkey not to expire" assert sign == subkey.can_sign
assert subkey.can_encrypt, \ assert encrypt == subkey.can_encrypt
"Default subkey capabilities do not include encryption" assert authenticate == subkey.can_authenticate
t = 2 * 24 * 60 * 60 # Check algorithm
slack = 5 * 60 res = ctx.create_subkey(key, algorithm="rsa")
res = ctx.create_subkey(key, expires_in=t) subkey = get_subkey(res.fpr)
subkey = get_subkey(res.fpr) assert subkey.pubkey_algo == 1
assert abs(time.time() + t - subkey.expires) < slack, \
"subkeys expiration time is off"
# Check capabilities # Check algorithm with size
for sign, encrypt, authenticate in itertools.product([False, True], res = ctx.create_subkey(key, algorithm="rsa1024")
[False, True], subkey = get_subkey(res.fpr)
[False, True]): assert subkey.pubkey_algo == 1
# Filter some out assert subkey.length == 1024
if not (sign or encrypt or authenticate):
# This triggers the default capabilities tested before.
continue
res = ctx.create_subkey(key, sign=sign, encrypt=encrypt, # Check algorithm future-default
authenticate=authenticate) ctx.create_subkey(key, algorithm="future-default")
subkey = get_subkey(res.fpr)
assert sign == subkey.can_sign
assert encrypt == subkey.can_encrypt
assert authenticate == subkey.can_authenticate
# Check algorithm # Check passphrase protection. For this we create a new key
res = ctx.create_subkey(key, algorithm="rsa") # so that we have a key with just one encryption subkey.
subkey = get_subkey(res.fpr) bravo_res = ctx.create_key(bravo, certify=True)
assert subkey.pubkey_algo == 1 bravo_key = ctx.get_key(bravo_res.fpr)
assert len(bravo_key.subkeys) == 1, "Expected one primary key and no subkeys"
# Check algorithm with size passphrase = "streng geheim"
res = ctx.create_subkey(key, algorithm="rsa1024") res = ctx.create_subkey(bravo_key, passphrase=passphrase)
subkey = get_subkey(res.fpr) ciphertext, _, _ = ctx.encrypt(b"hello there",
assert subkey.pubkey_algo == 1 recipients=[ctx.get_key(bravo_res.fpr)])
assert subkey.length == 1024
# Check algorithm future-default cb_called = False
ctx.create_subkey(key, algorithm="future-default") def cb(*args):
global cb_called
cb_called = True
return passphrase
ctx.pinentry_mode = gpg.constants.PINENTRY_MODE_LOOPBACK
ctx.set_passphrase_cb(cb)
# Check passphrase protection. For this we create a new key plaintext, _, _ = ctx.decrypt(ciphertext)
# so that we have a key with just one encryption subkey. assert plaintext == b"hello there"
bravo_res = ctx.create_key(bravo, certify=True) assert cb_called
bravo_key = ctx.get_key(bravo_res.fpr)
assert len(bravo_key.subkeys) == 1, "Expected one primary key and no subkeys"
passphrase = "streng geheim"
res = ctx.create_subkey(bravo_key, passphrase=passphrase)
ciphertext, _, _ = ctx.encrypt(b"hello there",
recipients=[ctx.get_key(bravo_res.fpr)])
cb_called = False
def cb(*args):
global cb_called
cb_called = True
return passphrase
ctx.pinentry_mode = gpg.constants.PINENTRY_MODE_LOOPBACK
ctx.set_passphrase_cb(cb)
plaintext, _, _ = ctx.decrypt(ciphertext)
assert plaintext == b"hello there"
assert cb_called