Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
lambda: Client.from_url(
reactor=reactor,
url=URL.from_text(acme_url),
key=load_or_create_client_key(account_key_file),
alg=RS256,
)
client_creator=(lambda: Client.from_url(
reactor=reactor,
url=LETSENCRYPT_DIRECTORY,
key=load_or_create_client_key(FilePath(certificates)),
alg=RS256,
)),
for issuing certs.
:param str pemdir: The path to the certificate directory to use.
"""
def colon_join(items):
return ':'.join([item.replace(':', '\\:') for item in items])
timeout = kwargs.pop('timeout', _DEFAULT_TIMEOUT)
sub = colon_join(list(args) + ['='.join(item) for item in kwargs.items()])
pem_path = FilePath(pemdir).asTextMode()
acme_key = load_or_create_client_key(pem_path)
return AutoTLSEndpoint(
reactor=reactor,
directory=directory,
client=Client.from_url(
reactor, directory, key=acme_key, alg=RS256, timeout=timeout),
cert_store=DirectoryStore(pem_path),
cert_mapping=HostDirectoryMap(pem_path),
sub_endpoint=serverFromString(reactor, sub))
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,
)
# Service to start.
WaitingResponder(
LibcloudDNSResponder.create(
reactor, driver_name, user_name, password, zone_name
),
reactor
)
]
acme_key = maybe_key(acme_path)
cert_store = DirectoryStore(acme_path)
if staging:
le_url = LETSENCRYPT_STAGING_DIRECTORY
else:
le_url = LETSENCRYPT_DIRECTORY
client_creator = partial(Client.from_url, reactor=reactor,
url=le_url,
key=acme_key, alg=RS256)
clock = reactor
service = AcmeIssuingService(cert_store, client_creator, clock,
responders)
service._registered = False
return service._check_certs()
return secretly(reactor, action=action,
def from_url(
cls, reactor, url, key, alg=RS256,
jws_client=None, timeout=_DEFAULT_TIMEOUT,
):
"""
Construct a client from an ACME directory at a given URL.
At construct time, it validates the ACME directory.
:param url: The ``twisted.python.url.URL`` to fetch the directory from.
See `txacme.urls` for constants for various well-known public
directories.
:param reactor: The Twisted reactor to use.
:param ~josepy.jwk.JWK key: The client key to use.
:param alg: The signing algorithm to use. Needs to be compatible with
the type of key used.
:param JWSClient jws_client: The underlying client to use, or ``None``
to construct one.
lambda: Client.from_url(
reactor=self.reactor,
url=URL.from_text(self.hs.config.acme_url),
key=load_or_create_client_key(
FilePath(self.hs.config.config_dir_path)
),
alg=RS256,
)
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,))
# We request a single certificate for a list of domains and get an "order"