This file is indexed.

/usr/share/pyshared/landscape/watchdog.py is in landscape-common 12.04.3-0ubuntu1.

This file is owned by root:root, with mode 0o644.

The actual contents of the file can be viewed below.

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
"""See L{WatchDog}.

The WatchDog must run as root, because it spawns the Landscape Manager.

The main C{landscape-client} program uses this watchdog.
"""

import os
import errno
import sys
import pwd
import signal
import time

from logging import warning, info, error
from resource import setrlimit, RLIMIT_NOFILE

from twisted.internet import reactor
from twisted.internet.defer import Deferred, succeed
from twisted.internet.protocol import ProcessProtocol
from twisted.internet.error import ProcessExitedAlready
from twisted.application.service import Service, Application
from twisted.application.app import startApplication

from landscape.deployment import init_logging, Configuration
from landscape.lib.twisted_util import gather_results
from landscape.lib.log import log_failure
from landscape.lib.bootstrap import (BootstrapList, BootstrapFile,
                                     BootstrapDirectory)
from landscape.log import rotate_logs
from landscape.amp import ComponentProtocol
from landscape.broker.amp import (
    RemoteBrokerConnector, RemoteMonitorConnector, RemoteManagerConnector)
from landscape.reactor import TwistedReactor
from landscape.lib.dns import discover_server
from landscape.configuration import (
    fetch_base64_ssl_public_certificate, decode_base64_ssl_public_certificate)

GRACEFUL_WAIT_PERIOD = 10
MAXIMUM_CONSECUTIVE_RESTARTS = 5
RESTART_BURST_DELAY = 30  # seconds
SIGKILL_DELAY = 10


class DaemonError(Exception):
    """One of the daemons could not be started."""


class TimeoutError(Exception):
    """Something took too long."""


class ExecutableNotFoundError(Exception):
    """An executable was not found."""


class Daemon(object):
    """A Landscape daemon which can be started and tracked.

    This class should be subclassed to specify individual daemon.

    @cvar program: The name of the executable program that will start this
        daemon.
    @cvar username: The name of the user to switch to, by default.
    @cvar service: The DBus service name that the program will be expected to
        listen on.
    @cvar max_retries: The maximum number of retries before giving up when
        trying to connect to the watched daemon.
    @cvar factor: The factor by which the delay between subsequent connection
        attempts will increase.

    @param connector: The L{RemoteComponentConnector} of the daemon.
    @param reactor: The reactor used to spawn the process and schedule timed
        calls.
    @param verbose: Optionally, report more information when running this
        program.  Defaults to False.
    """

    username = "landscape"
    max_retries = 3
    factor = 1.1
    options = None

    def __init__(self, connector, reactor=reactor, verbose=False,
                 config=None):
        self._connector = connector
        self._reactor = reactor
        self._env = os.environ.copy()
        if os.getuid() == 0:
            pwd_info = pwd.getpwnam(self.username)
            self._uid = pwd_info.pw_uid
            self._gid = pwd_info.pw_gid
            self._env["HOME"] = pwd_info.pw_dir
            self._env["USER"] = self.username
            self._env["LOGNAME"] = self.username
        else:
            # We can only switch UIDs if we're root, so simply don't switch
            # UIDs if we're not.
            self._uid = None
            self._gid = None
        self._verbose = verbose
        self._config = config
        self._process = None
        self._last_started = 0
        self._quick_starts = 0
        self._allow_restart = True

    def find_executable(self):
        """Find the fully-qualified path to the executable.

        If the executable can't be found, L{ExecutableNotFoundError} will be
        raised.
        """
        dirname = os.path.dirname(os.path.abspath(sys.argv[0]))
        executable = os.path.join(dirname, self.program)
        if not os.path.exists(executable):
            raise ExecutableNotFoundError("%s doesn't exist" % (executable,))
        return executable

    def start(self):
        """Start this daemon."""
        self._process = None

        now = time.time()
        if self._last_started + RESTART_BURST_DELAY > now:
            self._quick_starts += 1
            if self._quick_starts == MAXIMUM_CONSECUTIVE_RESTARTS:
                error("Can't keep %s running. Exiting." % self.program)
                self._reactor.stop()
                return
        else:
            self._quick_starts = 0

        self._last_started = now

        self._process = WatchedProcessProtocol(self)
        exe = self.find_executable()
        args = [exe, "--ignore-sigint"]
        if not self._verbose:
            args.append("--quiet")
        if self._config:
            args.extend(["-c", self._config])
        if self.options is not None:
            args.extend(self.options)
        self._reactor.spawnProcess(self._process, exe, args=args,
                                   env=self._env, uid=self._uid, gid=self._gid)

    def stop(self):
        """Stop this daemon."""
        if not self._process:
            return succeed(None)
        return self._process.kill()

    def _connect_and_call(self, name, *args, **kwargs):
        """Connect to the remote daemon over AMP and perform the given command.

        @param name: The name of the command to perform.
        @param args: Arguments list to be passed to the connect method
        @param kwargs: Keywords arguments to pass to the connect method.
        @return: A L{Deferred} resulting in C{True} if the command was
            successful or C{False} otherwise.
        @see: L{RemoteLandscapeComponentCreator.connect}.
        """

        def disconnect(ignored):
            self._connector.disconnect()
            return True

        connected = self._connector.connect(self.max_retries, self.factor,
                                            quiet=True)
        connected.addCallback(lambda remote: getattr(remote, name)())
        connected.addCallback(disconnect)
        connected.addErrback(lambda x: False)
        return connected

    def request_exit(self):
        return self._connect_and_call("exit")

    def is_running(self):
        # FIXME Error cases may not be handled in the best possible way
        # here. We're basically return False if any error happens from the
        # dbus ping.
        return self._connect_and_call("ping")

    def wait(self):
        """
        Return a Deferred which will fire when the process has died.
        """
        if not self._process:
            return succeed(None)
        return self._process.wait()

    def wait_or_die(self):
        """
        Wait for the process to die for C{GRACEFUL_WAIT_PERIOD}. If it hasn't
        died by that point, send it a SIGTERM. If it doesn't die for
        C{SIGKILL_DELAY},
        """
        if not self._process:
            return succeed(None)
        return self._process.wait_or_die()

    def prepare_for_shutdown(self):
        """Called by the watchdog when starting to shut us down.

        It will prevent our L{WatchedProcessProtocol} to restart the process
        when it exits.
        """
        self._allow_restart = False

    def allow_restart(self):
        """Return a boolean indicating if the daemon should be restarted."""
        return self._allow_restart

    def rotate_logs(self):
        self._process.rotate_logs()


class Broker(Daemon):
    program = "landscape-broker"


class Monitor(Daemon):
    program = "landscape-monitor"


class Manager(Daemon):
    program = "landscape-manager"
    username = "root"


class WatchedProcessProtocol(ProcessProtocol):
    """
    A process-watching protocol which sends any of its output to the log file
    and restarts it when it dies.
    """

    _killed = False

    def __init__(self, daemon):
        self.daemon = daemon
        self._wait_result = None
        self._delayed_really_kill = None
        self._delayed_terminate = None

    def kill(self):
        self._terminate()
        return self.wait()

    def _terminate(self, warn=False):
        if self.transport is not None:
            if warn:
                warning("%s didn't exit. Sending SIGTERM"
                        % (self.daemon.program,))
            try:
                self.transport.signalProcess(signal.SIGTERM)
            except ProcessExitedAlready:
                pass
            else:
                # Give some time for the process, and then show who's the boss.
                delayed = reactor.callLater(SIGKILL_DELAY, self._really_kill)
                self._delayed_really_kill = delayed

    def _really_kill(self):
        try:
            self.transport.signalProcess(signal.SIGKILL)
        except ProcessExitedAlready:
            pass
        else:
            warning("%s didn't die.  Sending SIGKILL." % self.daemon.program)
        self._delayed_really_kill = None

    def rotate_logs(self):
        if self.transport is not None:
            try:
                self.transport.signalProcess(signal.SIGUSR1)
            except ProcessExitedAlready:
                pass

    def wait(self):
        if self.transport.pid is None:
            return succeed(None)
        self._wait_result = Deferred()
        return self._wait_result

    def wait_or_die(self):
        self._delayed_terminate = reactor.callLater(GRACEFUL_WAIT_PERIOD,
                                                    self._terminate, warn=True)
        return self.wait()

    def outReceived(self, data):
        # it's *probably* going to always be line buffered, by accident
        sys.stdout.write(data)

    def errReceived(self, data):
        sys.stderr.write(data)

    def processEnded(self, reason):
        """The process has ended; restart it."""
        if self._delayed_really_kill is not None:
            self._delayed_really_kill.cancel()
        if (self._delayed_terminate is not None
            and self._delayed_terminate.active()):
            self._delayed_terminate.cancel()
        if self._wait_result is not None:
            self._wait_result.callback(None)
        elif self.daemon.allow_restart():
            self.daemon.start()


class WatchDog(object):
    """
    The Landscape WatchDog starts all other landscape daemons and ensures that
    they are working.
    """

    def __init__(self, reactor=reactor, verbose=False, config=None,
                 broker=None, monitor=None, manager=None,
                 enabled_daemons=None):
        twisted_reactor = TwistedReactor()
        if enabled_daemons is None:
            enabled_daemons = [Broker, Monitor, Manager]
        if broker is None and Broker in enabled_daemons:
            broker = Broker(
                RemoteBrokerConnector(twisted_reactor, config),
                verbose=verbose, config=config.config)
        if monitor is None and Monitor in enabled_daemons:
            monitor = Monitor(
                RemoteMonitorConnector(twisted_reactor, config),
                verbose=verbose, config=config.config)
        if manager is None and Manager in enabled_daemons:
            manager = Manager(
                RemoteManagerConnector(twisted_reactor, config),
                verbose=verbose, config=config.config)

        self.broker = broker
        self.monitor = monitor
        self.manager = manager
        self.daemons = filter(None, [self.broker, self.monitor, self.manager])
        self.reactor = reactor
        self._checking = None
        self._stopping = False
        signal.signal(signal.SIGUSR1, self._notify_rotate_logs)
        if config is not None and config.clones > 0:
            options = ["--clones", str(config.clones),
                       "--start-clones-over", str(config.start_clones_over)]
            for daemon in self.daemons:
                daemon.options = options

        self._ping_failures = {}

    def check_running(self):
        """Return a list of any daemons that are already running."""
        results = []
        for daemon in self.daemons:
            # This method is called on startup, we basically try to connect
            # a few times in fast sequence (with exponential backoff), if we
            # don't get a response we assume the daemon is not running.
            result = daemon.is_running()
            result.addCallback(lambda is_running, d=daemon: (is_running, d))
            results.append(result)

        def got_all_results(r):
            return [x[1] for x in r if x[0]]
        return gather_results(results).addCallback(got_all_results)

    def start(self):
        """
        Start all daemons. The broker will be started first, and no other
        daemons will be started before it is running and responding to DBUS
        messages.

        @return: A deferred which fires when all services have successfully
            started. If a daemon could not be started, the deferred will fail
            with L{DaemonError}.
        """
        for daemon in self.daemons:
            daemon.start()
        self.start_monitoring()

    def start_monitoring(self):
        """Start monitoring processes which have already been started."""
        # Must wait before daemons actually start, otherwise check will
        # restart them *again*.
        self._checking = self.reactor.callLater(5, self._check)

    def _restart_if_not_running(self, is_running, daemon):
        if (not is_running) and (not self._stopping):
            warning("%s failed to respond to a ping."
                    % (daemon.program,))
            if daemon not in self._ping_failures:
                self._ping_failures[daemon] = 0
            self._ping_failures[daemon] += 1
            if self._ping_failures[daemon] == 5:
                warning("%s died! Restarting." % (daemon.program,))
                stopping = daemon.stop()

                def stopped(ignored):
                    daemon.start()
                    self._ping_failures[daemon] = 0
                stopping.addBoth(stopped)
                return stopping
        else:
            self._ping_failures[daemon] = 0

    def _check(self):
        all_running = []
        for daemon in self.daemons:
            is_running = daemon.is_running()
            is_running.addCallback(self._restart_if_not_running, daemon)
            all_running.append(is_running)

        def reschedule(ignored):
            self._checking = self.reactor.callLater(5, self._check)
        gather_results(all_running).addBoth(reschedule)

    def request_exit(self):
        if self._checking is not None and self._checking.active():
            self._checking.cancel()
        # Set a flag so that the pinger will avoid restarting the daemons if a
        # ping has already been sent but not yet responded to.
        self._stopping = True

        # This tells the daemons to not automatically restart when they end
        for daemon in self.daemons:
            daemon.prepare_for_shutdown()

        def terminate_processes(broker_stopped):
            if broker_stopped:
                results = [daemon.wait_or_die() for daemon in self.daemons]
            else:
                # If request_exit fails, we should just kill the daemons
                # immediately.
                error("Couldn't request that broker gracefully shut down; "
                      "killing forcefully.")
                results = [x.stop() for x in self.daemons]
            return gather_results(results)

        result = self.broker.request_exit()
        return result.addCallback(terminate_processes)

    def _notify_rotate_logs(self, signal, frame):
        for daemon in self.daemons:
            daemon.rotate_logs()
        rotate_logs()


class WatchDogConfiguration(Configuration):

    def make_parser(self):
        parser = super(WatchDogConfiguration, self).make_parser()
        parser.add_option("--daemon", action="store_true",
                          help="Fork and run in the background.")
        parser.add_option("--pid-file", type="str",
                          help="The file to write the PID to.")
        parser.add_option("--monitor-only", action="store_true",
                          help="Don't enable management features. This is "
                          "useful if you want to run the client as a non-root "
                          "user.")
        return parser

    def get_enabled_daemons(self):
        daemons = [Broker, Monitor]
        if not self.monitor_only:
            daemons.append(Manager)
        return daemons


def daemonize():
    # See http://www.steve.org.uk/Reference/Unix/faq_2.html#SEC16
    if os.fork():  # launch child and...
        os._exit(0)  # kill off parent
    os.setsid()
    if os.fork():  # launch child and...
        os._exit(0)  # kill off parent again.
    # some argue that this umask should be 0, but that's annoying.
    os.umask(077)
    null = os.open('/dev/null', os.O_RDWR)
    for i in range(3):
        try:
            os.dup2(null, i)
        except OSError, e:
            if e.errno != errno.EBADF:
                raise
    os.close(null)


class WatchDogService(Service):

    def __init__(self, config):
        self._config = config
        self.watchdog = WatchDog(verbose=not config.daemon,
                                 config=config,
                                 enabled_daemons=config.get_enabled_daemons())
        self.exit_code = 0

    def autodiscover(self):
        """
        Autodiscover called if config setting config.server_autodiscover is
        True. This method allows the watchdog to attempt server autodiscovery,
        fetch the discovered landscape server's custom CA certificate, and
        write both the certificate and the updated config file with the
        discovered values.
        """
        def update_config(hostname):
            if hostname is None:
                warning("Autodiscovery returned empty hostname string. "
                        "Reverting to previous settings.")
            else:
                info("Autodiscovery found landscape server at %s. "
                     "Updating configuration values." % hostname)
                self._config.server_autodiscover = False
                self._config.url = "https://%s/message-system" % hostname
                self._config.ping_url = "http://%s/ping" % hostname
                if not self._config.ssl_public_key:
                    # If we don't have a key on this system, pull it from
                    # the auto-discovered server and write it to the filesystem
                    ssl_public_key = fetch_base64_ssl_public_certificate(
                        hostname, on_info=info, on_error=warning)
                    if ssl_public_key:
                        self._config.ssl_public_key = ssl_public_key
                        decode_base64_ssl_public_certificate(self._config)
                self._config.write()
            return hostname

        def discovery_error(result):
            warning("Autodiscovery failed.  Reverting to previous settings.")

        lookup_deferred = discover_server(
            self._config.autodiscover_srv_query_string,
            self._config.autodiscover_a_query_string)
        lookup_deferred.addCallback(update_config)
        lookup_deferred.addErrback(discovery_error)
        return lookup_deferred

    def startService(self):
        Service.startService(self)
        bootstrap_list.bootstrap(data_path=self._config.data_path,
                                 log_dir=self._config.log_dir)
        if self._config.clones > 0:

            # Let clones open an appropriate number of fds
            setrlimit(RLIMIT_NOFILE, (self._config.clones * 100,
                                      self._config.clones * 200))

            # Increase the timeout of AMP's MethodCalls
            ComponentProtocol.timeout = 300

            # Create clones log and data directories
            for i in range(self._config.clones):
                suffix = "-clone-%d" % i
                bootstrap_list.bootstrap(
                    data_path=self._config.data_path + suffix,
                    log_dir=self._config.log_dir + suffix)

        result = succeed(None)
        if self._config.server_autodiscover:
            result.addCallback(lambda _: self.autodiscover())
        result.addCallback(lambda _: self.watchdog.check_running())

        def start_if_not_running(running_daemons):
            if running_daemons:
                error("ERROR: The following daemons are already running: %s"
                      % (", ".join(x.program for x in running_daemons)))
                self.exit_code = 1
                reactor.crash()  # so stopService isn't called.
                return
            self._daemonize()
            info("Watchdog watching for daemons.")
            return self.watchdog.start()

        def die(failure):
            log_failure(failure, "Unknown error occurred!")
            self.exit_code = 2
            reactor.crash()
        result.addCallback(start_if_not_running)
        result.addErrback(die)
        return result

    def _daemonize(self):
        if self._config.daemon:
            daemonize()
            if self._config.pid_file:
                stream = open(self._config.pid_file, "w")
                stream.write(str(os.getpid()))
                stream.close()

    def stopService(self):
        info("Stopping client...")
        Service.stopService(self)

        # If CTRL-C is pressed twice in a row, the second SIGINT actually
        # kills us before subprocesses die, and that makes them hang around.
        signal.signal(signal.SIGINT, signal.SIG_IGN)

        done = self.watchdog.request_exit()
        done.addBoth(lambda r: self._remove_pid())
        return done

    def _remove_pid(self):
        pid_file = self._config.pid_file
        if pid_file is not None and os.access(pid_file, os.W_OK):
            stream = open(pid_file)
            pid = stream.read()
            stream.close()
            if pid == str(os.getpid()):
                os.unlink(pid_file)


bootstrap_list = BootstrapList([
    BootstrapDirectory("$data_path", "landscape", "root", 0755),
    BootstrapDirectory("$data_path/package", "landscape", "root", 0755),
    BootstrapDirectory(
        "$data_path/package/hash-id", "landscape", "root", 0755),
    BootstrapDirectory(
        "$data_path/package/binaries", "landscape", "root", 0755),
    BootstrapDirectory(
        "$data_path/package/upgrade-tool", "landscape", "root", 0755),
    BootstrapDirectory("$data_path/messages", "landscape", "root", 0755),
    BootstrapDirectory("$data_path/sockets", "landscape", "root", 0750),
    BootstrapDirectory(
        "$data_path/custom-graph-scripts", "landscape", "root", 0755),
    BootstrapDirectory("$log_dir", "landscape", "root", 0755),
    BootstrapFile("$data_path/package/database", "landscape", "root", 0644),
    ])


def clean_environment():
    """Unset dangerous environment variables.

    In particular unset all variables beginning with DEBIAN_ or DEBCONF_,
    to avoid any problems when landscape-client is invoked from its
    postinst script.  Some environment variables may be set which would affect
    *other* maintainer scripts which landscape-client invokes (via smart).
    """
    for key in os.environ.keys():
        if (key.startswith("DEBIAN_")
            or key.startswith("DEBCONF_")
            or key in ["LANDSCAPE_ATTACHMENTS", "MAIL"]):
            del os.environ[key]


def run(args=sys.argv, reactor=None):
    """Start the watchdog.

    This is the topmost function that kicks off the Landscape client.  It
    cleans up the environment, loads the configuration, and starts the
    reactor.

    @param args: Command line arguments, including the program name as the
        first element.
    @param reactor: The reactor to use.  If none is specified, the global
        reactor is used.
    @raise SystemExit: if command line arguments are bad, or when landscape-
        client is not running as 'root' or 'landscape'.
    """
    clean_environment()

    config = WatchDogConfiguration()
    config.load(args)

    try:
        landscape_uid = pwd.getpwnam("landscape").pw_uid
    except KeyError:
        sys.exit("The 'landscape' user doesn't exist!")

    if os.getuid() not in (0, landscape_uid):
        sys.exit("landscape-client can only be run as 'root' or 'landscape'.")

    init_logging(config, "watchdog")

    application = Application("landscape-client")
    watchdog_service = WatchDogService(config)
    watchdog_service.setServiceParent(application)

    if reactor is None:
        from twisted.internet import reactor
    # We add a small delay to work around a Twisted bug: this method should
    # only be called when the reactor is running, but we still get a
    # PotentialZombieWarning.
    reactor.callLater(0, startApplication, application, False)
    reactor.run()
    return watchdog_service.exit_code