Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def _test_create_client(self):
with start_action(action_type=u'integration:create_client').context():
self.key = JWKRSA(key=generate_private_key('rsa'))
return (
DeferredContext(self._create_client(self.key))
.addActionFinish())
to use, as with the endpoint.
"""
acme_key_file = pem_path.asTextMode().child(u'client.key')
if acme_key_file.exists():
key = serialization.load_pem_private_key(
acme_key_file.getContent(),
password=None,
backend=default_backend())
else:
key = generate_private_key(u'rsa')
acme_key_file.setContent(
key.private_bytes(
encoding=serialization.Encoding.PEM,
format=serialization.PrivateFormat.TraditionalOpenSSL,
encryption_algorithm=serialization.NoEncryption()))
return JWKRSA(key=key)
if acme_key_file.exists():
key = serialization.load_pem_private_key(
acme_key_file.getContent(),
password=None,
backend=default_backend()
)
else:
key = generate_private_key(u'rsa')
acme_key_file.setContent(
key.private_bytes(
encoding=serialization.Encoding.PEM,
format=serialization.PrivateFormat.TraditionalOpenSSL,
encryption_algorithm=serialization.NoEncryption()
)
)
acme_key = JWKRSA(key=key)
return acme_key
def get_things_done():
"""
Here is where the client part is setup and action is done.
"""
responders = yield start_responders()
# We first validate the directory.
account_key = _get_account_key()
try:
client = yield Client.from_url(
reactor,
URL.fromText(acme_url.decode('utf-8')),
key=JWKRSA(key=account_key),
alg=RS256,
)
except Exception as error:
print('\n\nFailed to connect to ACME directory. %s' % (error,))
yield reactor.stop()
defer.returnValue(None)
# Then we register a new account or update an existing account.
# First register a new account with a contact set, then using the same
# key call register with a different contact and see that it was updated.
response = yield client.start(
email='txacme-test1@twstedmatrix.org,txacme-test2@twstedmatrix.org')
print('Account URI: %s' % (response.uri,))
print('Account contact: %s' % (response.body.contact,))
def get_things_done():
"""
Here is where the service part is setup and action is done.
"""
responders = yield start_responders()
store = MemoryStore()
# We first validate the directory.
account_key = _get_account_key()
try:
client = yield Client.from_url(
reactor,
URL.fromText(acme_url.decode('utf-8')),
key=JWKRSA(key=account_key),
alg=RS256,
)
except Exception as error:
print('\n\nFailed to connect to ACME directory. %s' % (error,))
yield reactor.stop()
defer.returnValue(None)
service = AcmeIssuingService(
email='txacme-test1@twstedmatrix.org,txacme-test2@twstedmatrix.org',
cert_store=store,
client=client,
clock=reactor,
responders=responders,
panic=on_panic,
)