1#!/usr/bin/env python3

2"""

3 Program to test that ntpdate will sync the clock with an internet time server.

4

5 Copyright (C) 2010 Canonical Ltd.

6

7 Author:

8 Jeff Lane <jeffrey.lane@canonical.com>

9

10 This program is free software: you can redistribute it and/or modify

11 it under the terms of the GNU General Public License version 2,

12 as published by the Free Software Foundation.

13

14 This program is distributed in the hope that it will be useful,

15 but WITHOUT ANY WARRANTY; without even the implied warranty of

16 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the

17 GNU General Public License for more details.

18

19 You should have received a copy of the GNU General Public License

20 along with this program. If not, see <http://www.gnu.org/licenses/>.

21

22 The purpose of this script is to test to see whether the test system can

23 connect to an internet time server and sync the local clock.

24

25 It will also check to see if ntpd is running locally, and if so, stop it for

26 the duration of the test and restart it after the test is finished.

27

28 By default, we're hitting ntp.ubuntu.com, however you can use any valid NTP

29 server by passing the URL to the program via --server

30

31"""

32 import sys

33 import os

34 import logging

35 import signal

36 import time

37 from datetime import datetime, timedelta

38 from subprocess import Popen, PIPE

39 from argparse import ArgumentParser

40

41

42 def SilentCall(*popenargs):

43 """

44 Modified version of subprocess.call() to supress output from the command

45 that is executed. Wait for command to complete, then return the returncode

46 attribute.

47 """

48 null_fh = open("/dev/null", "wb", 0)

49 try:

50 return Popen(

51 *popenargs, shell=True, stdout=null_fh, stderr=null_fh

52 ).wait()

53 finally:

54 null_fh.close()

55

56

57 def CheckNTPD():

58 """

59 This checks to see if nptd is running or not, if so it returns a tuple

60 (status,pid,command) where status is either on or off.

61 """

62 ps_list = (

63 Popen("ps axo pid,comm", shell=True, stdout=PIPE)

64 .communicate()[0]

65 .splitlines()

66 )

67 for item in ps_list:

68 fields = item.split()

69 if fields[1] == "ntpd":

70 logging.debug("Found %s with PID %s" % (fields[1], fields[0]))

71 break

72 if fields[1] == "ntpd":

73 return ("on", fields[0], fields[1])

74 else:

75 return ("off", "0", "0")

76

77

78 def StartStopNTPD(state, pid=0):

79 """

80 This is used to either start or stop ntpd if its running.

81 """

82 if state == "off":

83 logging.info("Stopping ntpd process PID: %s" % pid)

84 os.kill(int(pid), signal.SIGHUP)

85 elif state == "on":

86 logging.info("Starting ntp process")

87 SilentCall("/etc/init.d/ntp start")

88 ntpd_status = CheckNTPD()

89

90 if ntpd_status == 0:

91 logging.debug("ntpd restarted with PID: %s" % ntpd_status[1])

92 else:

93 logging.error("ntpd restart failed for some reason")

94 else:

95 logging.error(

96 "%s is an unknown state, unable to start/stop ntpd" % state

97 )

98

99

100 def SyncTime(server):

101 """

102 This is used to sync time to the specified ntp server. We use -b here as

103 that syncs time faster than the slewed method that ntpdate uses by default,

104 meaning we'll see something meaningful faster.

105 """

106 cmd = "ntpdate -b " + server

107 logging.debug("using %s" % server)

108 sync = Popen(cmd, shell=True, stdout=PIPE, stderr=PIPE)

109 result = sync.communicate()

110

111 if sync.returncode == 0:

112 logging.info("Successful NTP update from %s" % server)

113 logging.debug(result[0].strip().decode())

114 return True

115 else:

116 logging.error("Failed to sync with %s" % server)

117 logging.error(result[1].strip().decode())

118 return False

119

120

121 def TimeCheck():

122 """

123 Returns current time in a time.localtime() struct

124 """

125 return time.localtime()

126

127

128 def SkewTime():

129 """

130 Optional function. We can skew time by 1 hour if we'd like to see real sync

131 changes being enforced

132 """

133 TIME_SKEW = 1

134 logging.info("Time Skewing has been selected. Setting clock ahead 1 hour")

135 # Let's get our current time

136 skewed = datetime.now() + timedelta(hours=TIME_SKEW)

137 logging.info("Current time is: %s" % time.asctime())

138 # Now create new time string in the form MMDDhhmmYYYY for the date program

139 date_time_string = skewed.strftime("%m%d%H%M%Y")

140 logging.debug("New date string is: %s" % date_time_string)

141 logging.debug("Setting new system time/date")

142 # This call is necessary for testing, otherwise TimeSkew() does nothing.

143 SilentCall("/bin/date %s" % date_time_string)

144 logging.info("Pre-sync time is: %s" % time.asctime())

145

146

147 def main():

148 description = (

149 "Tests the ability to skew and sync the clock with an NTP server"

150 )

151 parser = ArgumentParser(description=description)

152 parser.add_argument(

153 "--server",

154 action="store",

155 default="ntp.ubuntu.com",

156 help="The NTP server to sync from. The default server \

157 is %(default)s",

158 )

159 parser.add_argument(

160 "--skew-time",

161 action="store_true",

162 default=False,

163 help="Setting this will change system time ahead by 1 \

164 hour to make the results of ntp syncing more dramatic \

165 and noticeable.",

166 )

167 parser.add_argument(

168 "-d",

169 "--debug",

170 action="store_true",

171 default=False,

172 help="Verbose output for debugging purposes",

173 )

174 args = parser.parse_args()

175

176 if os.geteuid() != 0:

177 parser.error("You must run this script as root")

178

179 # Set up logging

180 format = "%(asctime)s %(levelname)-8s %(message)s"

181 date_format = "%Y-%m-%d %H:%M:%S"

182 handler = logging.StreamHandler()

183 handler.setFormatter(logging.Formatter(format, date_format))

184 logger = logging.getLogger()

185 logger.setLevel(logging.INFO)

186 if args.debug:

187 logger.setLevel(logging.DEBUG)

188 logger.addHandler(handler)

189

190 # Make sure NTP is installed

191 if not os.access("/usr/sbin/ntpdate", os.F_OK):

192 logging.error("NTP is not installed!")

193 return 1

194

195 # Check for and stop the ntp daemon

196 ntpd_status = CheckNTPD()

197 logging.debug(

198 "Pre-sync ntpd status: %s %s %s"

199 % (ntpd_status[0], ntpd_status[1], ntpd_status[2])

200 )

201 if ntpd_status[0] == "on":

202 logging.debug("Since ntpd is currently running, stopping it now")

203 StartStopNTPD("off", ntpd_status[1])

204

205 if args.skew_time:

206 logging.debug("Setting system time ahead one hour")

207 SkewTime()

208 else:

209 logging.info("Pre-sync time is: %s" % time.asctime(TimeCheck()))

210

211 sync = SyncTime(args.server)

212

213 logging.info("Current system time is: %s" % time.asctime(TimeCheck()))

214

215 # Restart ntp daemon

216 if ntpd_status[0] == "on":

217 logging.debug("Since ntpd was previously running, starting it again")

218 StartStopNTPD("on")

219

220 if sync is True:

221 return 0

222 else:

223 return 1

224

225

226 if __name__ == "__main__":

227 sys.exit(main())