Source code for ldaptor.generate_password
from twisted.internet import process, protocol, defer
from twisted.python import failure
[docs]class PwgenException(Exception):
pass
[docs]class ReadPassword(protocol.ProcessProtocol):
def __init__(self, deferred, count=1):
self.deferred = deferred
self.count = count
self.stdout = ""
self.stderr = ""
[docs] def outReceived(self, data):
self.stdout = self.stdout + data
[docs] def errReceived(self, data):
self.stderr = self.stderr + data
[docs] def processEnded(self, reason):
if self.stderr:
self.deferred.errback(failure.Failure(PwgenException(reason, self.stderr)))
elif self.stdout:
lines = [x for x in self.stdout.split("\n") if x]
if len(lines) != self.count:
self.deferred.errback(
failure.Failure(
PwgenException(reason, "Wrong number of lines received.")
)
)
self.deferred.callback(lines)
else:
self.deferred.errback(failure.Failure(PwgenException(reason, "")))
[docs]def generate(reactor, n=1):
assert n > 0
d = defer.Deferred()
proto = ReadPassword(d, n)
process.Process(
reactor, "pwgen", ("pwgen", "-cn1", "-N", "%d" % n), {}, None, proto
)
return d
if __name__ == "__main__":
from twisted.internet import reactor
import sys
def say(passwords):
for p in passwords:
sys.stdout.write("%s\n" % p)
return passwords
def err(fail):
fail.trap(PwgenException)
sys.stderr.write("pwgen: %s\n" % fail.getErrorMessage())
# Could get more passwords in one fork, but this stresses it more
# on purpose.
l = []
for i in range(5):
d = generate(reactor, 5)
d.addCallbacks(say, err)
l.append(d)
dl = defer.DeferredList(l)
dl.addBoth(lambda dummy: reactor.stop())
reactor.run()