monkey.py 3.2 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990
  1. import os, sys
  2. from tempfile import NamedTemporaryFile
  3. from subprocess import Popen
  4. from fuse import FUSE, Operations, LoggingMixIn
  5. try:
  6. from urllib.request import urlopen # Python 3
  7. except ImportError:
  8. from urllib2 import urlopen # Python 2
  9. # domain with server.py running on it for testing
  10. DOMAIN = os.getenv("TRAVIS_DOMAIN", "travis-ci.gethttpsforfree.com")
  11. # generate account and domain keys
  12. def gen_keys():
  13. # good account key
  14. account_key = NamedTemporaryFile()
  15. Popen(["openssl", "genrsa", "-out", account_key.name, "2048"]).wait()
  16. # weak 1024 bit key
  17. weak_key = NamedTemporaryFile()
  18. Popen(["openssl", "genrsa", "-out", weak_key.name, "1024"]).wait()
  19. # good domain key
  20. domain_key = NamedTemporaryFile()
  21. domain_csr = NamedTemporaryFile()
  22. Popen(["openssl", "req", "-newkey", "rsa:2048", "-nodes", "-keyout", domain_key.name,
  23. "-subj", "/CN={0}".format(DOMAIN), "-out", domain_csr.name]).wait()
  24. # subject alt-name domain
  25. san_csr = NamedTemporaryFile()
  26. san_conf = NamedTemporaryFile()
  27. for openssl_cnf in ['/etc/pki/tls/openssl.cnf', '/etc/ssl/openssl.cnf']:
  28. if os.path.exists(openssl_cnf): break
  29. san_conf.write(open(openssl_cnf).read().encode("utf8"))
  30. san_conf.write("\n[SAN]\nsubjectAltName=DNS:{0}\n".format(DOMAIN).encode("utf8"))
  31. san_conf.seek(0)
  32. Popen(["openssl", "req", "-new", "-sha256", "-key", domain_key.name,
  33. "-subj", "/", "-reqexts", "SAN", "-config", san_conf.name,
  34. "-out", san_csr.name]).wait()
  35. # invalid domain csr
  36. invalid_csr = NamedTemporaryFile()
  37. Popen(["openssl", "req", "-new", "-sha256", "-key", domain_key.name,
  38. "-subj", "/CN=\xC3\xA0\xC2\xB2\xC2\xA0_\xC3\xA0\xC2\xB2\xC2\xA0.com", "-out", invalid_csr.name]).wait()
  39. # nonexistent domain csr
  40. nonexistent_csr = NamedTemporaryFile()
  41. Popen(["openssl", "req", "-new", "-sha256", "-key", domain_key.name,
  42. "-subj", "/CN=404.gethttpsforfree.com", "-out", nonexistent_csr.name]).wait()
  43. # account-signed domain csr
  44. account_csr = NamedTemporaryFile()
  45. Popen(["openssl", "req", "-new", "-sha256", "-key", account_key.name,
  46. "-subj", "/CN={0}".format(DOMAIN), "-out", account_csr.name]).wait()
  47. return {
  48. "account_key": account_key,
  49. "weak_key": weak_key,
  50. "domain_key": domain_key,
  51. "domain_csr": domain_csr,
  52. "san_csr": san_csr,
  53. "invalid_csr": invalid_csr,
  54. "nonexistent_csr": nonexistent_csr,
  55. "account_csr": account_csr,
  56. }
  57. # fake a folder structure to catch the key authorization file
  58. FS = {}
  59. class Passthrough(LoggingMixIn, Operations): # pragma: no cover
  60. def getattr(self, path, fh=None):
  61. f = FS.get(path, None)
  62. if f is None:
  63. return super(Passthrough, self).getattr(path, fh=fh)
  64. return f
  65. def write(self, path, buf, offset, fh):
  66. urlopen("http://{0}/.well-known/acme-challenge/?{1}".format(DOMAIN,
  67. os.getenv("TRAVIS_SESSION", "not_set")), buf)
  68. return len(buf)
  69. def create(self, path, mode, fi=None):
  70. FS[path] = {"st_mode": 33204}
  71. return 0
  72. def unlink(self, path):
  73. del(FS[path])
  74. return 0
  75. if __name__ == "__main__": # pragma: no cover
  76. FUSE(Passthrough(), sys.argv[1], nothreads=True, foreground=True)