fake_ntp_server.py 2.2 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677
  1. #!/usr/bin/env python
  2. from __future__ import print_function
  3. import argparse
  4. import socket
  5. import time
  6. from ntp_packet import (
  7. NTPPacket,
  8. NTPTimestamp,
  9. )
  10. def fake_server(address, dateunixtime):
  11. s = socket.socket(socket.AF_INET6, socket.SOCK_DGRAM)
  12. s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
  13. s.bind( (address, 123) )
  14. drift_start = time.time()
  15. speed = 0.9996
  16. while True:
  17. packet, source = s.recvfrom(100)
  18. print('Received {length} byte packet from {source}'.format(length=len(packet), source=source[0]))
  19. packet = NTPPacket.from_bytes(packet)
  20. if packet.version not in (3, 4):
  21. print('Unsupported NTP version.')
  22. continue
  23. if packet.mode != 3:
  24. print('Not a client request.')
  25. continue
  26. try:
  27. print('Client time:', packet.transmit_timestamp)
  28. except:
  29. pass
  30. time_since_start = dateunixtime - drift_start
  31. return_time = drift_start + time_since_start * speed
  32. response = NTPPacket()
  33. response.version = packet.version
  34. response.mode = 4
  35. response.stratum = 1
  36. response.reference_identifier = b'XFAK'
  37. response.origin_timestamp = packet.transmit_timestamp
  38. response.reference_timestamp = NTPTimestamp.from_unix_timestamp(return_time)
  39. response.receive_timestamp = NTPTimestamp.from_unix_timestamp(return_time)
  40. response.transmit_timestamp = NTPTimestamp.from_unix_timestamp(return_time)
  41. response = response.to_bytes()
  42. s.sendto(response, source)
  43. def parse_bind_addr(address):
  44. address = socket.getaddrinfo(address, 123, 0, 0, 0, socket.AI_NUMERICHOST)
  45. address = address[0]
  46. if address[0] == socket.AF_INET:
  47. return '::ffff:' + address[4][0]
  48. else:
  49. return address[4][0]
  50. def parse_args():
  51. parser = argparse.ArgumentParser(description='Fake NTP server')
  52. parser.add_argument('--listen', type=parse_bind_addr, default='::')
  53. parser.add_argument('--unixtime', default='1')
  54. return parser.parse_args()
  55. def main():
  56. args = parse_args()
  57. fake_server(args.listen, args.unixtime)
  58. if __name__ == '__main__':
  59. main()