1
# Copyright 2014 Canonical Ltd. This software is licensed under the
2
# GNU Affero General Public License version 3 (see the file LICENSE).
4
"""Twisted Application Plugin code for the MAAS Region."""
6
from __future__ import (
16
"IntrospectionShellService",
20
from twisted.application.internet import StreamServerEndpointService
21
from twisted.conch import manhole
22
from twisted.conch.insults import insults
23
from twisted.internet import (
27
from twisted.internet.interfaces import IProtocolFactory
28
from zope.interface import implementer
31
def byteString(thing):
32
"""Convert `string` to a byte string."""
33
if isinstance(thing, bytes):
35
elif isinstance(thing, unicode):
36
return thing.encode("utf-8")
39
"Cannot safely convert %r to a byte string."
43
def serverFromString(description):
44
"""Parse `description` into an endpoint.
46
Twisted's `endpoints.serverFromString` raises errors that are not
47
particularly user-friendly, so we attempt something better here.
49
description = byteString(description)
51
return endpoints.serverFromString(reactor, description)
54
"Could not understand server description %r. "
55
"Try something like %r or %r." % (
56
description, b"unix:/path/to/file:mode=660:lockfile=0",
57
b"tcp:1234:interface=127.0.0.1"))
61
"""Define the builtin 'help'.
63
This is a wrapper around pydoc.help... with a twist, in that it disallows
68
message = "Type help(object) for help about object."
69
return message.encode("utf-8")
71
def __call__(self, *args, **kwds):
73
message = "Interactive help has been disabled. %r" % self
74
print(message.encode("utf-8"))
76
from pydoc import help
77
return help(*args, **kwds)
80
class IntrospectionShell(manhole.ColoredManhole):
82
STYLE_BRIGHT_ON = '\x1b[1m'
85
COLOUR_RED_ON = '\x1b[31m'
86
COLOUR_MAGENTA_ON = '\x1b[35m'
87
COLOUR_OFF = '\x1b[39m'
89
def __init__(self, namespace=None):
90
super(IntrospectionShell, self).__init__(namespace)
91
self.ensureThereIsHelp()
93
def ensureThereIsHelp(self):
94
if self.namespace is None:
95
self.namespace = {"help": Help()}
96
elif "help" in self.namespace:
97
pass # Don't override.
99
self.namespace["help"] = Help()
101
def welcomeMessage(self):
103
self.STYLE_BRIGHT_ON,
104
self.COLOUR_MAGENTA_ON,
105
"Welcome to MAAS's Introspection Shell.",
109
self.STYLE_BRIGHT_ON,
112
self.factory.location.upper(),
118
def initializeScreen(self):
119
"""Override in order to provide welcome message."""
120
self.terminal.reset()
121
self.addOutput(b"\n")
122
self.addOutput(self.welcomeMessage().encode("utf-8"))
123
self.addOutput(b"\n")
124
self.addOutput(b"\n")
129
@implementer(IProtocolFactory)
130
class IntrospectionShellFactory:
132
def __init__(self, location, namespace):
133
super(IntrospectionShellFactory, self).__init__()
134
self.namespace = namespace
135
self.location = location
137
def buildProtocol(self, addr):
138
proto = insults.ServerProtocol(
139
IntrospectionShell, self.namespace)
144
"""See `IProtocolFactory`."""
147
"""See `IProtocolFactory`."""
150
class IntrospectionShellService(StreamServerEndpointService):
152
def __init__(self, location, endpoint, namespace):
153
factory = IntrospectionShellFactory(location, namespace=namespace)
154
super(IntrospectionShellService, self).__init__(endpoint, factory)