Merge lp:~allenap/maas/more-better-things--bug-1389007 into lp:~maas-committers/maas/trunk

[Gavin Panella]

Adapt AMP to use 32-bit length prefixes for on-wire serialization.

This allows us to have much larger arguments and responses over RPC, though it's limited to 2MiB for now, up from 64kiB.

This is a *big* branch, I know. There's no good way of doing this in stages. I copied the AMP code and tests from Twisted, adapted them for 32-bit-length-prefixed strings, then fixed up the code to pass muster in MAAS's coding style, investigated a test isolation failure, and then also adapted the tests to run using MAASTestCase instead of Twisted's Trial code.

I think the best way to verify/test/review this code is in CI and use. I've not changed the behaviour of the AMP code except the switch from 16 to 32 bit, so that can be reviewed solely for style, rather than substance. I did have to write AMPTestCase for the tests, but again most of the behaviour is intact.

The bulk of the rest is mechanical, which can be skipped through fairly quickly.

Please review this?

[Christian Reis (kiko)]

Hmm, do we really want to copy the code wholesale instead of changing AMT upstream to be fixed, parametrizable or monkey-patchable?

review: Needs Information

[Christian Reis (kiko)]

AMT on my mind, I obviously menat AMP.

[Gavin Panella (allenap)]

There's no way we can change upstream AMP to 32-bit prefixes; it's an incompatible change. We could propose amp32 as an additional protocol, or we could write negotiation code that starts on 16-bit and switches to 32-bit if supported.

The former would be fine: it's something we can try though it may be rejected, but it doesn't block this branch.

The latter is, at a guess, several days work instead of ~6 hours that went into this. I'm okay to do it, but I guess we've got more important things to do.

[Jeroen T. Vermeulen (jtv)]

Not a full review yet, but some notes:

.

Typo: "This contain the".

.

If you're going to include Twisted code here, it may not be valid to say "See LICENSE for details." You may need to include their licence file under a recognisable name, and make the copyright header refer to it as such. Also, if you're creating a derived work with substantial changes, I guess it needs our copyright notice as well.

.

AMPTestCase (not AMP32TestCase?) and everything in it needs docstrings.

Although to be honest, I'd prefer not to have yet another test-case class in the first place! I see very little in there that actually needs to be in a class: shortDescription might be generic or it could be in a fixture. setUp *is* a fixture. getLoggedFailures could be in MAASTestCase. And assertWarns is a generic matcher disguised as a specific helper which makes non-transparent decisions about whether the test should continue after failure.

.

In AMPTestCase.assertWarns, if you know you've got a list of one item and you want the item, prefer extracting it using pattern-matching ("[item] = mylist") over indexing ("mylist[0]").

[Jeroen T. Vermeulen (jtv)]

More notes:

.

It'd be worth noting in the documentation (or did I miss it?) that this is a modified version of the Twisted implementation, and that it does not change the on-the-wire protocol

.

I'd stick the module-level verifyClass invocations in tests.

.

I wasn't sure about AMPTestCase, but the AMPTest class almost certainly should say AMP32 instead of plain AMP.

.

For the bulk of the code, I can only assume that it's as good as the code we've been relying on anyway. So I'm voting Approve for the code as such — though I'll leave it to others (including Jenkins) to figure out how well the landing would fit into the grand plan.

review: Approve

[lp:~allenap/maas/more-better-things--bug-1389007]

3372. By Gavin Panella on 2014-11-12

Fix typo.

3373. By Gavin Panella on 2014-11-12

Include Twisted's LICENSE file which is references by amp32 and its tests.

3374. By Gavin Panella on 2014-11-12

shortDescription() is no longer needed.

3375. By Gavin Panella on 2014-11-12

Rename AMPTestCase to AMP32TestCase.

3376. By Gavin Panella on 2014-11-12

Docstrings for AMP32TestCase.

3377. By Gavin Panella on 2014-11-12

Use the pattern match, Luke.

[Gavin Panella (allenap)]

> Typo: "This contain the".

Well spotted, thanks. Fixed.

>
> .
>
> If you're going to include Twisted code here, it may not be valid to
> say "See LICENSE for details." You may need to include their licence
> file under a recognisable name, and make the copyright header refer to
> it as such. Also, if you're creating a derived work with substantial
> changes, I guess it needs our copyright notice as well.

I've pulled Twisted's LICENSE file in now, as LICENSE.Twisted, and
updated the headers in those files. Canonical is already listed in
LICENSE.Twisted so I think we can probably leave it as is.

>
> .
>
> AMPTestCase (not AMP32TestCase?) and everything in it needs
> docstrings.
>
> Although to be honest, I'd prefer not to have yet another test-case
> class in the first place! I see very little in there that actually
> needs to be in a class: shortDescription might be generic or it could
> be in a fixture. setUp *is* a fixture. getLoggedFailures could be in
> MAASTestCase. And assertWarns is a generic matcher disguised as a
> specific helper which makes non- transparent decisions about whether
> the test should continue after failure.

shortDescription() is a unittest method that gets called before the test
runs, but now that I've rebased on MAASTestCase instead of Trial's
TestCase it can go.

I've added docstrings which explain the other functions, and justify
their continued existence. I could spend time on getting rid of
AMP32TestCase, but the pay off doesn't seem worth it right now. I'm
biased, but I think it's already an improvement over the tests that work
with Trial.

>
> .
>
> In AMPTestCase.assertWarns, if you know you've got a list of one item
> and you want the item, prefer extracting it using pattern-matching
> ("[item] = mylist") over indexing ("mylist[0]").

Done.

Thanks!

[lp:~allenap/maas/more-better-things--bug-1389007]

3378. By Gavin Panella on 2014-11-12

Update amp32's docstring to draw attention to the differences from Twisted's amp.

3379. By Gavin Panella on 2014-11-12

Draw more attention to the difference between amp and amp32 in the docs.

3380. By Gavin Panella on 2014-11-12

Move module-level interface tests into a test case.

3381. By Gavin Panella on 2014-11-12

Rename AMPTest to AMP32Test.

[Gavin Panella (allenap)]

> It'd be worth noting in the documentation (or did I miss it?) that
> this is a modified version of the Twisted implementation, and that it
> does not change the on-the-wire protocol

Done.

>
> .
>
> I'd stick the module-level verifyClass invocations in tests.

Done.

>
> .
>
> I wasn't sure about AMPTestCase, but the AMPTest class almost
> certainly should say AMP32 instead of plain AMP.

Done.

>
> .
>
> For the bulk of the code, I can only assume that it's as good as the
> code we've been relying on anyway. So I'm voting Approve for the code
> as such — though I'll leave it to others (including Jenkins) to figure
> out how well the landing would fit into the grand plan.

Yeah, it's the same code, modified to our style conventions. The tests
have been modified to use testtools instead of Trial, but otherwise
they're the same.

Thanks again!

[Christian Reis (kiko)]

I think this needs a discussion upstream in order to establish the best path forward.

If upstream is willing to accept an upgrade-to-32bit patch, then we should invest energy into that because long-term it's much more sustainable for us. If they have another suggested solution we should use it.

We should only really take this fork in if there's really no other viable solution, which should be quite unlikely.

[Gavin Panella (allenap)]

> I think this needs a discussion upstream in order to establish the
> best path forward.
>
> If upstream is willing to accept an upgrade-to-32bit patch, then we
> should invest energy into that because long-term it's much more
> sustainable for us. If they have another suggested solution we should
> use it.
>
> We should only really take this fork in if there's really no other
> viable solution, which should be quite unlikely.

I think it's the right and proper thing to do to talk to the Twisted
project, and see if this can go upstream.

Twisted is a community project; we or I would likely be the de facto
maintainers.

Sustainably, my guess is that we're not in for a lot of work in either
case. AMP in Twisted is essentially done. This branch doesn't add new
features, just a change to an on-the-wire detail.

Even so, being upstream would give others the chance to use the code,
and give greater visibility to bugs and security issues.

Anecdotally, though from my own experience of the Twisted development
process, I can say that it's onerous to get a patch reviewed and landed.

If we get it upstream it could be a long time before that release makes
it back into Ubuntu. It's unlikely to get into the current LTS. We'd
need to carry this code anyway, for a while.

In all, I think we can make it a goal to get this upstream, but I don't
think we should block on it, except just to check there isn't a
ready-made alternative.

[Christian Reis (kiko)]

Yes, if we do agree that we want this patch and that it can go upstream,
then we should carry this in-tree until it is in a released version.

Please let us know what the immediate reaction upstream is and we'll
figure out what to do; I assume as you do that there's no obvious other
solution, but it would be nice to be surprised.
--
Christian Robottom Reis | [+1] 612 888 4935 | http://launchpad.net/~kiko
Canonical VP Hyperscale | [+55 16] 9 9112 6430 | http://async.com.br/~kiko

[Gavin Panella (allenap)]

The conversation with upstream developer exarkun [1] has put me off this branch. Not in a never-ever sense, but in the sense that I think this 64k limit is actually making us think. We can actually improve the power-poller (the trigger for this branch) by working within this limitation.

[1] http://twistedmatrix.com/pipermail/twisted-python/2014-November/028941.html

[Julian Edwards (julian-edwards)]

On Friday 14 Nov 2014 12:05:28 you wrote:
> The conversation with upstream developer exarkun [1] has put me off this
> branch. Not in a never-ever sense, but in the sense that I think this 64k
> limit is actually making us think. We can actually improve the power-poller
> (the trigger for this branch) by working within this limitation.
>
> [1]
> http://twistedmatrix.com/pipermail/twisted-python/2014-November/028941.html

I saw your mention of a "priority queue". I'm not sure we need that sort of
solution; we talked today about only querying the oldest 10 (say) power
statuses at a time. This will fit easily in the message size.

[Gavin Panella (allenap)]

> I saw your mention of a "priority queue". I'm not sure we need that
> sort of solution; we talked today about only querying the oldest 10
> (say) power statuses at a time. This will fit easily in the message
> size.

I was using big words to mean just that :)

[MAAS Lander (maas-lander)]

Transitioned to Git.

lp:maas has now moved from Bzr to Git.
Please propose your branches with Launchpad using Git.

git clone https://git.launchpad.net/maas

Unmerged revisions

3381. By Gavin Panella on 2014-11-12

Rename AMPTest to AMP32Test.

3380. By Gavin Panella on 2014-11-12

Move module-level interface tests into a test case.

3379. By Gavin Panella on 2014-11-12

Draw more attention to the difference between amp and amp32 in the docs.

3378. By Gavin Panella on 2014-11-12

Update amp32's docstring to draw attention to the differences from Twisted's amp.

3377. By Gavin Panella on 2014-11-12

Use the pattern match, Luke.

3376. By Gavin Panella on 2014-11-12

Docstrings for AMP32TestCase.

3375. By Gavin Panella on 2014-11-12

Rename AMPTestCase to AMP32TestCase.

3374. By Gavin Panella on 2014-11-12

shortDescription() is no longer needed.

3373. By Gavin Panella on 2014-11-12

Include Twisted's LICENSE file which is references by amp32 and its tests.

3372. By Gavin Panella on 2014-11-12

Fix typo.

=== added file 'LICENSE.Twisted'
--- LICENSE.Twisted	1970-01-01 00:00:00 +0000
+++ LICENSE.Twisted	2014-11-12 12:20:32 +0000
@@ -0,0 +1,65 @@
+Copyright (c) 2001-2014
+Allen Short
+Andy Gayton
+Andrew Bennetts
+Antoine Pitrou
+Apple Computer, Inc.
+Ashwini Oruganti
+Benjamin Bruheim
+Bob Ippolito
+Canonical Limited
+Christopher Armstrong
+David Reid
+Donovan Preston
+Eric Mangold
+Eyal Lotem
+Google Inc.
+Hybrid Logic Ltd.
+Hynek Schlawack
+Itamar Turner-Trauring
+James Knight
+Jason A. Mobarak
+Jean-Paul Calderone
+Jessica McKellar
+Jonathan Jacobs
+Jonathan Lange
+Jonathan D. Simms
+Jürgen Hermann
+Julian Berman
+Kevin Horn
+Kevin Turner
+Laurens Van Houtven
+Mary Gardiner
+Matthew Lefkowitz
+Massachusetts Institute of Technology
+Moshe Zadka
+Paul Swartz
+Pavel Pergamenshchik
+Ralph Meijer
+Richard Wall
+Sean Riley
+Software Freedom Conservancy
+Travis B. Hartwell
+Thijs Triemstra
+Thomas Herve
+Timothy Allen
+Tom Prince
+
+Permission is hereby granted, free of charge, to any person obtaining
+a copy of this software and associated documentation files (the
+"Software"), to deal in the Software without restriction, including
+without limitation the rights to use, copy, modify, merge, publish,
+distribute, sublicense, and/or sell copies of the Software, and to
+permit persons to whom the Software is furnished to do so, subject to
+the following conditions:
+
+The above copyright notice and this permission notice shall be
+included in all copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
+EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
+MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
+NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE
+LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
+OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
+WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 
=== modified file 'docs/development/rpc.rst'
--- docs/development/rpc.rst	2014-08-21 20:10:47 +0000
+++ docs/development/rpc.rst	2014-11-12 12:20:32 +0000
@@ -5,7 +5,11 @@
 
 MAAS contains an RPC mechanism such that every process in the region is
 connected to every process in the cluster (strictly, every pserv
-process). It's based on AMP_, specifically `Twisted's implementation`_.
+process). It's based on AMP_, specifically `Twisted's implementation`_,
+but MAAS incorporates an extended version that can transmit much larger
+messages by using 32-bit length prefixes in place of 16-bit. This makes
+it on-the-wire incompatible with standard AMP_ implementations, but its
+essential behaviour is identical.
 
 .. _AMP:
   http://amp-protocol.net/
@@ -17,22 +21,22 @@
 Where do I start?
 -----------------
 
-Start in the :py:mod:`provisioningserver.rpc` package. The first two files to
-look at are ``cluster.py`` and ``region.py``. This contain the
+Start in the :py:mod:`provisioningserver.rpc` package. The first two
+files to look at are ``cluster.py`` and ``region.py``. This contains the
 declarations of what commands are available on clusters and regions
 respectively.
 
 A new command could be declared like so::
 
-  from twisted.protocols import amp
+  from provisioningserver.rpc import amp32
 
-  class EatCheez(amp.Command):
+  class EatCheez(amp32.Command):
       arguments = [
-          (b"name", amp.Unicode()),
-          (b"origin", amp.Unicode()),
+          (b"name", amp32.Unicode()),
+          (b"origin", amp32.Unicode()),
       ]
       response = [
-          (b"rating", amp.Integer()),
+          (b"rating", amp32.Integer()),
       ]
 
 It's also possible to map exceptions across the wire using an ``errors``
 
=== modified file 'scripts/ampclient.py'
--- scripts/ampclient.py	2014-01-31 16:15:00 +0000
+++ scripts/ampclient.py	2014-11-12 12:20:32 +0000
@@ -16,6 +16,7 @@
 
 import sys
 
+from provisioningserver.rpc.amp32 import AMP
 from provisioningserver.rpc.cluster import ListBootImages
 from provisioningserver.rpc.region import ReportBootImages
 from twisted.internet import reactor
@@ -23,7 +24,6 @@
     connectProtocol,
     TCP4ClientEndpoint,
     )
-from twisted.protocols.amp import AMP
 
 
 def callRemote(command, port, **kwargs):
 
=== modified file 'src/maasserver/clusterrpc/power.py'
--- src/maasserver/clusterrpc/power.py	2014-09-16 21:00:33 +0000
+++ src/maasserver/clusterrpc/power.py	2014-11-12 12:20:32 +0000
@@ -38,7 +38,7 @@
     Nodes can be in any cluster; the power calls will be directed to their
     owning cluster.
 
-    :param command: The `amp.Command` to call.
+    :param command: The `amp32.Command` to call.
     :param nodes: A sequence of ``(system-id, hostname, cluster-uuid,
         power-info)`` tuples.
     :returns: A mapping of each node's system ID to a
 
=== modified file 'src/maasserver/clusterrpc/utils.py'
--- src/maasserver/clusterrpc/utils.py	2014-10-08 09:43:51 +0000
+++ src/maasserver/clusterrpc/utils.py	2014-11-12 12:20:32 +0000
@@ -38,7 +38,7 @@
 
     :param nodegroups: The :class:`NodeGroup`s on which to make the RPC
         call. If None, defaults to all :class:`NodeGroup`s.
-    :param command: An :class:`amp.Command` to call on the clusters.
+    :param command: An :class:`amp32.Command` to call on the clusters.
     :param ignore_errors: If True, errors encountered whilst calling
         `command` on the clusters won't raise an exception.
     :return: A generator of results, i.e. the dicts returned by the RPC
 
=== modified file 'src/maasserver/models/node.py'
--- src/maasserver/models/node.py	2014-11-10 03:22:59 +0000
+++ src/maasserver/models/node.py	2014-11-12 12:20:32 +0000
@@ -111,6 +111,7 @@
 from piston.models import Token
 from provisioningserver.logger import get_maas_logger
 from provisioningserver.power.poweraction import UnknownPowerType
+from provisioningserver.rpc import amp32
 from provisioningserver.rpc.cluster import (
     CancelMonitor,
     StartMonitors,
@@ -118,7 +119,6 @@
 from provisioningserver.rpc.power import QUERY_POWER_TYPES
 from provisioningserver.utils.enum import map_enum_reverse
 from provisioningserver.utils.twisted import asynchronous
-from twisted.protocols import amp
 
 
 maaslog = get_maas_logger("node")
@@ -584,7 +584,7 @@
             'node_status': self.status,
             'timeout': timeout,
             }
-        deadline = datetime.now(tz=amp.utc) + timedelta(seconds=timeout)
+        deadline = datetime.now(tz=amp32.utc) + timedelta(seconds=timeout)
         monitors = [{
             'deadline': deadline,
             'id': self.system_id,
 
=== modified file 'src/maasserver/models/tests/test_node.py'
--- src/maasserver/models/tests/test_node.py	2014-11-10 03:22:53 +0000
+++ src/maasserver/models/tests/test_node.py	2014-11-12 12:20:32 +0000
@@ -101,7 +101,10 @@
     )
 from provisioningserver.power.poweraction import UnknownPowerType
 from provisioningserver.power_schema import JSON_POWER_TYPE_PARAMETERS
-from provisioningserver.rpc import cluster as cluster_module
+from provisioningserver.rpc import (
+    amp32,
+    cluster as cluster_module,
+    )
 from provisioningserver.rpc.cluster import StartMonitors
 from provisioningserver.rpc.exceptions import NoConnectionsAvailable
 from provisioningserver.rpc.power import QUERY_POWER_TYPES
@@ -114,7 +117,6 @@
     )
 from twisted.internet import defer
 from twisted.internet.defer import Deferred
-from twisted.protocols import amp
 from twisted.python.failure import Failure
 
 
@@ -2123,7 +2125,7 @@
 
     def test__start_transition_monitor_starts_monitor(self):
         rpc_fixture = self.prepare_rpc()
-        now = datetime.now(tz=amp.utc)
+        now = datetime.now(tz=amp32.utc)
         self.patch_datetime_now(now)
         node = factory.make_Node()
         cluster = rpc_fixture.makeCluster(node.nodegroup, StartMonitors)
@@ -2142,7 +2144,7 @@
         )
 
     def test__start_transition_monitor_copes_with_timeouterror(self):
-        now = datetime.now(tz=amp.utc)
+        now = datetime.now(tz=amp32.utc)
         self.patch_datetime_now(now)
         node = factory.make_Node()
         mock_client = Mock()
 
=== modified file 'src/maasserver/rpc/regionservice.py'
--- src/maasserver/rpc/regionservice.py	2014-11-10 15:11:58 +0000
+++ src/maasserver/rpc/regionservice.py	2014-11-12 12:20:32 +0000
@@ -57,6 +57,7 @@
 from maasserver.utils.async import transactional
 from netaddr import IPAddress
 from provisioningserver.rpc import (
+    amp32,
     cluster,
     common,
     exceptions,
@@ -86,7 +87,6 @@
 from twisted.internet.error import ConnectionClosed
 from twisted.internet.protocol import Factory
 from twisted.internet.threads import deferToThread
-from twisted.protocols import amp
 from twisted.python import log
 from zope.interface import implementer
 
@@ -151,11 +151,11 @@
         """
         return deferToThread(leases.update_leases, uuid, mappings)
 
-    @amp.StartTLS.responder
+    @amp32.StartTLS.responder
     def get_tls_parameters(self):
         """get_tls_parameters()
 
-        Implementation of :py:class:`~twisted.protocols.amp.StartTLS`.
+        Implementation of :py:class:`~provisioningserver.rpc.amp32.StartTLS`.
         """
         try:
             from provisioningserver.rpc.testing import tls
 
=== modified file 'src/maasserver/rpc/testing/fixtures.py'
--- src/maasserver/rpc/testing/fixtures.py	2014-10-08 21:59:23 +0000
+++ src/maasserver/rpc/testing/fixtures.py	2014-11-12 12:20:32 +0000
@@ -191,7 +191,7 @@
     def addCluster(self, protocol):
         """Add a new stub cluster using the given `protocol`.
 
-        The `protocol` should be an instance of `amp.AMP`.
+        The `protocol` should be an instance of `amp32.AMP`.
 
         :returns: py:class:`twisted.test.iosim.IOPump`
         """
@@ -315,7 +315,7 @@
     def addCluster(self, protocol):
         """Add a new stub cluster using the given `protocol`.
 
-        The `protocol` should be an instance of `amp.AMP`.
+        The `protocol` should be an instance of `amp32.AMP`.
 
         :returns: A `Deferred` that fires with the connected protocol
             instance.
 
=== modified file 'src/maasserver/rpc/tests/test_regionservice.py'
--- src/maasserver/rpc/tests/test_regionservice.py	2014-11-10 15:11:58 +0000
+++ src/maasserver/rpc/tests/test_regionservice.py	2014-11-12 12:20:32 +0000
@@ -87,6 +87,7 @@
 import netaddr
 from provisioningserver.network import discover_networks
 from provisioningserver.rpc import (
+    amp32,
     cluster,
     common,
     exceptions,
@@ -163,7 +164,6 @@
 from twisted.internet.interfaces import IStreamServerEndpoint
 from twisted.internet.protocol import Factory
 from twisted.internet.threads import deferToThread
-from twisted.protocols import amp
 from twisted.python import log
 from twisted.python.failure import Failure
 from zope.interface.verify import verifyObject
@@ -284,7 +284,7 @@
 
     def test_StartTLS_is_registered(self):
         protocol = Region()
-        responder = protocol.locateResponder(amp.StartTLS.commandName)
+        responder = protocol.locateResponder(amp32.StartTLS.commandName)
         self.assertIsNotNone(responder)
 
     def test_get_tls_parameters_returns_parameters(self):
@@ -292,7 +292,7 @@
         # However, locateResponder() returns a closure, so we have to
         # side-step it.
         protocol = Region()
-        cls, func = protocol._commandDispatch[amp.StartTLS.commandName]
+        cls, func = protocol._commandDispatch[amp32.StartTLS.commandName]
         self.assertThat(func(protocol), are_valid_tls_parameters)
 
     @wait_for_reactor
@@ -303,7 +303,7 @@
         # travelling over the wire as part of an AMP message. However,
         # the responder is not aware of this, and is called just like
         # any other.
-        d = call_responder(Region(), amp.StartTLS, {})
+        d = call_responder(Region(), amp32.StartTLS, {})
 
         def check(response):
             self.assertEqual({}, response)
@@ -1119,7 +1119,7 @@
         service.factory.protocol = HandshakingRegionServer
         protocol = service.factory.buildProtocol(addr=None)  # addr is unused.
         protocol.connectionMade()
-        connectionLost_up_call = self.patch(amp.AMP, "connectionLost")
+        connectionLost_up_call = self.patch(amp32.AMP, "connectionLost")
         self.assertDictEqual(
             {protocol.ident: {protocol}},
             service.connections)
 
=== modified file 'src/provisioningserver/pserv_services/dhcp_probe_service.py'
--- src/provisioningserver/pserv_services/dhcp_probe_service.py	2014-10-02 11:53:59 +0000
+++ src/provisioningserver/pserv_services/dhcp_probe_service.py	2014-11-12 12:20:32 +0000
@@ -22,6 +22,7 @@
 
 from provisioningserver.dhcp.detect import probe_interface
 from provisioningserver.logger.log import get_maas_logger
+from provisioningserver.rpc.amp32 import UnhandledCommand
 from provisioningserver.rpc.exceptions import NoConnectionsAvailable
 from provisioningserver.rpc.region import (
     GetClusterInterfaces,
@@ -37,7 +38,6 @@
     returnValue,
     )
 from twisted.internet.threads import deferToThread
-from twisted.protocols.amp import UnhandledCommand
 
 
 maaslog = get_maas_logger("dhcp.probe")
 
=== added file 'src/provisioningserver/rpc/amp32.py'
--- src/provisioningserver/rpc/amp32.py	1970-01-01 00:00:00 +0000
+++ src/provisioningserver/rpc/amp32.py	2014-11-12 12:20:32 +0000
@@ -0,0 +1,2646 @@
+# Copyright (c) 2005 Divmod, Inc.
+# Copyright (c) Twisted Matrix Laboratories.
+# See LICENSE.Twisted for details.
+
+"""
+This module implements a modified version of AMP, the Asynchronous Messaging
+Protocol, which is referred to as AMP32 from now on where it's necessary to
+differentiate between the two.
+
+The original AMP uses 16-bit length prefixes when sending messages, which
+limits payloads to 64k. This version uses 32-bit prefixes so that messages can
+be much larger, though in practice they're limit them to something much
+smaller that the maximum. Unfortunately this makes AMP32 incompatible with
+AMP, though their essential behaviour is identical.
+
+AMP is a protocol for sending multiple asynchronous request/response pairs over
+the same connection.  Requests and responses are both collections of key/value
+pairs.
+
+AMP is a very simple protocol which is not an application.  This module is a
+"protocol construction kit" of sorts; it attempts to be the simplest wire-level
+implementation of Deferreds.  AMP provides the following base-level features:
+
+    - Asynchronous request/response handling (hence the name)
+
+    - Requests and responses are both key/value pairs
+
+    - Binary transfer of all data: all data is length-prefixed.  Your
+      application will never need to worry about quoting.
+
+    - Command dispatching (like HTTP Verbs): the protocol is extensible, and
+      multiple AMP sub-protocols can be grouped together easily.
+
+The protocol implementation also provides a few additional features which are
+not part of the core wire protocol, but are nevertheless very useful:
+
+    - Tight TLS integration, with an included StartTLS command.
+
+    - Handshaking to other protocols: because AMP has well-defined message
+      boundaries and maintains all incoming and outgoing requests for you, you
+      can start a connection over AMP and then switch to another protocol.
+      This makes it ideal for firewall-traversal applications where you may
+      have only one forwarded port but multiple applications that want to use
+      it.
+
+Using AMP with Twisted is simple.  Each message is a command, with a response.
+You begin by defining a command type.  Commands specify their input and output
+in terms of the types that they expect to see in the request and response
+key-value pairs.  Here's an example of a command that adds two integers, 'a'
+and 'b'::
+
+    class Sum(amp32.Command):
+        arguments = [('a', amp32.Integer()),
+                     ('b', amp32.Integer())]
+        response = [('total', amp32.Integer())]
+
+Once you have specified a command, you need to make it part of a protocol, and
+define a responder for it.  Here's a 'JustSum' protocol that includes a
+responder for our 'Sum' command::
+
+    class JustSum(amp32.AMP):
+        def sum(self, a, b):
+            total = a + b
+            print 'Did a sum: %d + %d = %d' % (a, b, total)
+            return {'total': total}
+        Sum.responder(sum)
+
+Later, when you want to actually do a sum, the following expression will return
+a L{Deferred} which will fire with the result::
+
+    ClientCreator(reactor, amp32.AMP).connectTCP(...).addCallback(
+        lambda p: p.callRemote(Sum, a=13, b=81)).addCallback(
+            lambda result: result['total'])
+
+Command responders may also return Deferreds, causing the response to be
+sent only once the Deferred fires::
+
+    class DelayedSum(amp32.AMP):
+        def slowSum(self, a, b):
+            total = a + b
+            result = defer.Deferred()
+            reactor.callLater(3, result.callback, {'total': total})
+            return result
+        Sum.responder(slowSum)
+
+This is transparent to the caller.
+
+You can also define the propagation of specific errors in AMP.  For example,
+for the slightly more complicated case of division, we might have to deal with
+division by zero::
+
+    class Divide(amp32.Command):
+        arguments = [('numerator', amp32.Integer()),
+                     ('denominator', amp32.Integer())]
+        response = [('result', amp32.Float())]
+        errors = {ZeroDivisionError: 'ZERO_DIVISION'}
+
+The 'errors' mapping here tells AMP that if a responder to Divide emits a
+L{ZeroDivisionError}, then the other side should be informed that an error of
+the type 'ZERO_DIVISION' has occurred.  Writing a responder which takes
+advantage of this is very simple - just raise your exception normally::
+
+    class JustDivide(amp32.AMP):
+        def divide(self, numerator, denominator):
+            result = numerator / denominator
+            print 'Divided: %d / %d = %d' % (numerator, denominator, total)
+            return {'result': result}
+        Divide.responder(divide)
+
+On the client side, the errors mapping will be used to determine what the
+'ZERO_DIVISION' error means, and translated into an asynchronous exception,
+which can be handled normally as any L{Deferred} would be::
+
+    def trapZero(result):
+        result.trap(ZeroDivisionError)
+        print "Divided by zero: returning INF"
+        return 1e1000
+    ClientCreator(reactor, amp32.AMP).connectTCP(...).addCallback(
+        lambda p: p.callRemote(Divide, numerator=1234,
+                               denominator=0)
+        ).addErrback(trapZero)
+
+For a complete, runnable example of both of these commands, see the files in
+the Twisted repository::
+
+    doc/core/examples/ampserver.py
+    doc/core/examples/ampclient.py
+
+On the wire, AMP is a protocol which uses 2-byte lengths to prefix keys and
+values, and empty keys to separate messages::
+
+    <2-byte length><key><2-byte length><value>
+    <2-byte length><key><2-byte length><value>
+    ...
+    <2-byte length><key><2-byte length><value>
+    <NUL><NUL>                  # Empty Key == End of Message
+
+And so on.  Because it's tedious to refer to lengths and NULs constantly, the
+documentation will refer to packets as if they were newline delimited, like
+so::
+
+    C: _command: sum
+    C: _ask: ef639e5c892ccb54
+    C: a: 13
+    C: b: 81
+
+    S: _answer: ef639e5c892ccb54
+    S: total: 94
+
+Notes:
+
+In general, the order of keys is arbitrary.  Specific uses of AMP may impose an
+ordering requirement, but unless this is specified explicitly, any ordering may
+be generated and any ordering must be accepted.  This applies to the
+command-related keys I{_command} and I{_ask} as well as any other keys.
+
+Values are limited to the maximum encodable size in a 32-bit length.
+
+Keys are limited to the maximum encodable size in a 8-bit length, 255 bytes.
+Note that we still use 2-byte lengths to encode keys.  This small redundancy
+has several features:
+
+    - If an implementation becomes confused and starts emitting corrupt data,
+      or gets keys confused with values, many common errors will be signalled
+      immediately instead of delivering obviously corrupt packets.
+
+    - A single NUL will separate every key, and a double NUL separates
+      messages.  This provides some redundancy when debugging traffic dumps.
+
+    - NULs will be present at regular intervals along the protocol, providing
+      some padding for otherwise braindead C implementations of the protocol,
+      so that <stdio.h> string functions will see the NUL and stop.
+
+    - This makes it possible to run an AMP server on a port also used by a
+      plain-text protocol, and easily distinguish between non-AMP clients (like
+      web browsers) which issue non-NUL as the first byte, and AMP clients,
+      which always issue NUL as the first byte.
+
+"""
+
+from __future__ import (
+    absolute_import,
+    print_function,
+    # unicode_literals,
+    )
+
+str = None
+
+__metaclass__ = type
+
+import datetime
+import decimal
+from io import BytesIO
+from itertools import count
+from struct import pack
+import types
+import warnings
+
+from twisted.internet.defer import (
+    Deferred,
+    fail,
+    maybeDeferred,
+    )
+from twisted.internet.error import (
+    ConnectionClosed,
+    ConnectionLost,
+    PeerVerifyError,
+    )
+from twisted.internet.interfaces import IFileDescriptorReceiver
+from twisted.internet.main import CONNECTION_LOST
+from twisted.protocols.basic import (
+    Int32StringReceiver,
+    StatefulStringProtocol,
+    )
+from twisted.python import (
+    filepath,
+    log,
+    )
+from twisted.python.failure import Failure
+from twisted.python.reflect import accumulateClassDict
+from zope.interface import (
+    implements,
+    Interface,
+    )
+
+
+try:
+    from twisted.internet import ssl
+except ImportError:
+    ssl = None
+else:
+    if ssl.supported:
+        from twisted.internet.ssl import (
+            CertificateOptions,
+            Certificate,
+            DN,
+            KeyPair,
+        )
+    else:
+        ssl = None
+
+
+ASK = '_ask'
+ANSWER = '_answer'
+COMMAND = '_command'
+ERROR = '_error'
+ERROR_CODE = '_error_code'
+ERROR_DESCRIPTION = '_error_description'
+UNKNOWN_ERROR_CODE = 'UNKNOWN'
+UNHANDLED_ERROR_CODE = 'UNHANDLED'
+
+MAX_KEY_LENGTH = 0xff
+MAX_VALUE_LENGTH = 2 * 1024 * 1024  # 2MiB
+
+
+class IArgumentType(Interface):
+    """
+    An L{IArgumentType} can serialize a Python object into an AMP box and
+    deserialize information from an AMP box back into a Python object.
+
+    @since: 9.0
+    """
+
+    def fromBox(name, strings, objects, proto):
+        """
+        Given an argument name and an AMP box containing serialized values,
+        extract one or more Python objects and add them to the C{objects}
+        dictionary.
+
+        @param name: The name associated with this argument.  Most commonly,
+            this is the key which can be used to find a serialized value in
+            C{strings} and which should be used as the key in C{objects} to
+            associate with a structured Python object.
+        @type name: C{bytes}
+
+        @param strings: The AMP box from which to extract one or more
+            values.
+        @type strings: C{dict}
+
+        @param objects: The output dictionary to populate with the value for
+            this argument.
+        @type objects: C{dict}
+
+        @param proto: The protocol instance which received the AMP box being
+            interpreted.  Most likely this is an instance of L{AMP}, but
+            this is not guaranteed.
+
+        @return: C{None}
+        """
+
+    def toBox(name, strings, objects, proto):
+        """
+        Given an argument name and a dictionary containing structured Python
+        objects, serialize values into one or more strings and add them to
+        the C{strings} dictionary.
+
+        @param name: The name associated with this argument.  Most commonly,
+            this is the key which can be used to find an object in
+            C{objects} and which should be used as the key in C{strings} to
+            associate with a C{bytes} giving the serialized form of that
+            object.
+        @type name: C{bytes}
+
+        @param strings: The AMP box into which to insert one or more
+            strings.
+        @type strings: C{dict}
+
+        @param objects: The input dictionary from which to extract Python
+            objects to serialize.
+        @type objects: C{dict}
+
+        @param proto: The protocol instance which will send the AMP box once
+            it is fully populated.  Most likely this is an instance of
+            L{AMP}, but this is not guaranteed.
+
+        @return: C{None}
+        """
+
+
+class IBoxSender(Interface):
+    """
+    A transport which can send L{AmpBox} objects.
+    """
+
+    def sendBox(box):
+        """
+        Send an L{AmpBox}.
+
+        @raise ProtocolSwitched: if the underlying protocol has been
+        switched.
+
+        @raise ConnectionLost: if the underlying connection has already been
+        lost.
+        """
+
+    def unhandledError(failure):
+        """
+        An unhandled error occurred in response to a box.  Log it
+        appropriately.
+
+        @param failure: a L{Failure} describing the error that occurred.
+        """
+
+
+class IBoxReceiver(Interface):
+    """
+    An application object which can receive L{AmpBox} objects and dispatch them
+    appropriately.
+    """
+
+    def startReceivingBoxes(boxSender):
+        """
+        The L{ampBoxReceived} method will start being called; boxes may be
+        responded to by responding to the given L{IBoxSender}.
+
+        @param boxSender: an L{IBoxSender} provider.
+        """
+
+    def ampBoxReceived(box):
+        """
+        A box was received from the transport; dispatch it appropriately.
+        """
+
+    def stopReceivingBoxes(reason):
+        """
+        No further boxes will be received on this connection.
+
+        @type reason: L{Failure}
+        """
+
+
+class IResponderLocator(Interface):
+    """
+    An application object which can look up appropriate responder methods for
+    AMP commands.
+    """
+
+    def locateResponder(name):
+        """
+        Locate a responder method appropriate for the named command.
+
+        @param name: the wire-level name (commandName) of the AMP command to be
+        responded to.
+
+        @return: a 1-argument callable that takes an L{AmpBox} with argument
+        values for the given command, and returns an L{AmpBox} containing
+        argument values for the named command, or a L{Deferred} that fires the
+        same.
+        """
+
+
+class AmpError(Exception):
+    """
+    Base class of all Amp-related exceptions.
+    """
+
+
+class ProtocolSwitched(Exception):
+    """
+    Connections which have been switched to other protocols can no longer
+    accept traffic at the AMP level.  This is raised when you try to send it.
+    """
+
+
+class OnlyOneTLS(AmpError):
+    """
+    This is an implementation limitation; TLS may only be started once per
+    connection.
+    """
+
+
+class NoEmptyBoxes(AmpError):
+    """
+    You can't have empty boxes on the connection.  This is raised when you
+    receive or attempt to send one.
+    """
+
+
+class InvalidSignature(AmpError):
+    """
+    You didn't pass all the required arguments.
+    """
+
+
+class TooLong(AmpError):
+    """
+    One of the protocol's length limitations was violated.
+
+    @ivar isKey: true if the string being encoded in a key position, false if
+    it was in a value position.
+
+    @ivar isLocal: Was the string encoded locally, or received too long from
+    the network?  (It's only physically possible to encode "too long" values on
+    the network for keys.)
+
+    @ivar value: The string that was too long.
+
+    @ivar keyName: If the string being encoded was in a value position, what
+    key was it being encoded for?
+    """
+
+    def __init__(self, isKey, isLocal, value, keyName=None):
+        AmpError.__init__(self)
+        self.isKey = isKey
+        self.isLocal = isLocal
+        self.value = value
+        self.keyName = keyName
+
+    def __repr__(self):
+        hdr = self.isKey and "key" or "value"
+        if not self.isKey:
+            hdr += ' ' + repr(self.keyName)
+        lcl = self.isLocal and "local" or "remote"
+        return "%s %s too long: %d" % (lcl, hdr, len(self.value))
+
+
+class BadLocalReturn(AmpError):
+    """
+    A bad value was returned from a local command; we were unable to coerce it.
+    """
+
+    def __init__(self, message, enclosed):
+        AmpError.__init__(self)
+        self.message = message
+        self.enclosed = enclosed
+
+    def __repr__(self):
+        return self.message + " " + self.enclosed.getBriefTraceback()
+
+    __bytes__ = __repr__
+
+
+class RemoteAmpError(AmpError):
+    """
+    This error indicates that something went wrong on the remote end of the
+    connection, and the error was serialized and transmitted to you.
+    """
+
+    def __init__(self, errorCode, description, fatal=False, local=None):
+        """Create a remote error with an error code and description.
+
+        @param errorCode: the AMP error code of this error.
+
+        @param description: some text to show to the user.
+
+        @param fatal: a boolean, true if this error should terminate the
+        connection.
+
+        @param local: a local Failure, if one exists.
+        """
+        if local:
+            localwhat = ' (local)'
+            othertb = local.getBriefTraceback()
+        else:
+            localwhat = ''
+            othertb = ''
+        Exception.__init__(
+            self, "Code<%s>%s: %s%s" % (
+                errorCode, localwhat,
+                description, othertb))
+        self.local = local
+        self.errorCode = errorCode
+        self.description = description
+        self.fatal = fatal
+
+
+class UnknownRemoteError(RemoteAmpError):
+    """
+    This means that an error whose type we can't identify was raised from the
+    other side.
+    """
+
+    def __init__(self, description):
+        errorCode = UNKNOWN_ERROR_CODE
+        RemoteAmpError.__init__(self, errorCode, description)
+
+
+class MalformedAmpBox(AmpError):
+    """
+    This error indicates that the wire-level protocol was malformed.
+    """
+
+
+class UnhandledCommand(AmpError):
+    """
+    A command received via amp could not be dispatched.
+    """
+
+
+class IncompatibleVersions(AmpError):
+    """
+    It was impossible to negotiate a compatible version of the protocol with
+    the other end of the connection.
+    """
+
+
+PROTOCOL_ERRORS = {UNHANDLED_ERROR_CODE: UnhandledCommand}
+
+
+class AmpBox(dict):
+    """
+    I am a packet in the AMP protocol, much like a regular bytes:bytes
+    dictionary.
+    """
+
+    # be like a regular dictionary, don't magically acquire a __dict__...
+    __slots__ = []
+
+    def copy(self):
+        """
+        Return another AmpBox just like me.
+        """
+        newBox = self.__class__()
+        newBox.update(self)
+        return newBox
+
+    def serialize(self):
+        """
+        Convert me into a wire-encoded string.
+
+        @return: a bytes encoded according to the rules described in the module
+        docstring.
+        """
+        i = sorted(self.viewitems())
+        L = []
+        w = L.append
+        for k, v in i:
+            if type(k) == unicode:
+                raise TypeError("Unicode key not allowed: %r" % k)
+            if type(v) == unicode:
+                raise TypeError(
+                    "Unicode value for key %r not allowed: %r" % (k, v))
+            if len(k) > MAX_KEY_LENGTH:
+                raise TooLong(True, True, k, None)
+            if len(v) > MAX_VALUE_LENGTH:
+                raise TooLong(False, True, v, k)
+            for kv in k, v:
+                w(pack("!I", len(kv)))
+                w(kv)
+        w(pack("!I", 0))
+        return ''.join(L)
+
+    def _sendTo(self, proto):
+        """
+        Serialize and send this box to a Amp instance.  By the time it is being
+        sent, several keys are required.  I must have exactly ONE of::
+
+            _ask
+            _answer
+            _error
+
+        If the '_ask' key is set, then the '_command' key must also be
+        set.
+
+        @param proto: an AMP instance.
+        """
+        proto.sendBox(self)
+
+    def __repr__(self):
+        return 'AmpBox(%s)' % (dict.__repr__(self),)
+
+
+# amp32.Box => AmpBox
+Box = AmpBox
+
+
+class QuitBox(AmpBox):
+    """
+    I am an AmpBox that, upon being sent, terminates the connection.
+    """
+
+    __slots__ = []
+
+    def __repr__(self):
+        return 'QuitBox(**%s)' % (super(QuitBox, self).__repr__(),)
+
+    def _sendTo(self, proto):
+        """
+        Immediately call loseConnection after sending.
+        """
+        super(QuitBox, self)._sendTo(proto)
+        proto.transport.loseConnection()
+
+
+class _SwitchBox(AmpBox):
+    """
+    Implementation detail of ProtocolSwitchCommand: I am a AmpBox which sets
+    up state for the protocol to switch.
+    """
+
+    # DON'T set __slots__ here; we do have an attribute.
+
+    def __init__(self, innerProto, **kw):
+        """
+        Create a _SwitchBox with the protocol to switch to after being sent.
+
+        @param innerProto: the protocol instance to switch to.
+        @type innerProto: an IProtocol provider.
+        """
+        super(_SwitchBox, self).__init__(**kw)
+        self.innerProto = innerProto
+
+    def __repr__(self):
+        return '_SwitchBox(%r, **%s)' % (self.innerProto,
+                                         dict.__repr__(self),)
+
+    def _sendTo(self, proto):
+        """
+        Send me; I am the last box on the connection. All further traffic will
+        be over the new protocol.
+        """
+        super(_SwitchBox, self)._sendTo(proto)
+        proto._lockForSwitch()
+        proto._switchTo(self.innerProto)
+
+
+class BoxDispatcher:
+    """
+    A L{BoxDispatcher} dispatches '_ask', '_answer', and '_error' L{AmpBox}es,
+    both incoming and outgoing, to their appropriate destinations.
+
+    Outgoing commands are converted into L{Deferred}s and outgoing boxes, and
+    associated tracking state to fire those L{Deferred} when '_answer' boxes
+    come back.  Incoming '_answer' and '_error' boxes are converted into
+    callbacks and errbacks on those L{Deferred}s, respectively.
+
+    Incoming '_ask' boxes are converted into method calls on a supplied method
+    locator.
+
+    @ivar _outstandingRequests: a dictionary mapping request IDs to
+    L{Deferred}s which were returned for those requests.
+
+    @ivar locator: an object with a L{locateResponder} method that locates a
+    responder function that takes a Box and returns a result (either a Box or a
+    Deferred which fires one).
+
+    @ivar boxSender: an object which can send boxes, via the L{_sendBox}
+    method, such as an L{AMP} instance.
+    @type boxSender: L{IBoxSender}
+    """
+
+    implements(IBoxReceiver)
+
+    _failAllReason = None
+    _outstandingRequests = None
+    _counter = 0
+    boxSender = None
+
+    def __init__(self, locator):
+        self._outstandingRequests = {}
+        self.locator = locator
+
+    def startReceivingBoxes(self, boxSender):
+        """
+        The given boxSender is going to start calling boxReceived on this
+        L{BoxDispatcher}.
+
+        @param boxSender: The L{IBoxSender} to send command responses to.
+        """
+        self.boxSender = boxSender
+
+    def stopReceivingBoxes(self, reason):
+        """
+        No further boxes will be received here.  Terminate all currently
+        oustanding command deferreds with the given reason.
+        """
+        self.failAllOutgoing(reason)
+
+    def failAllOutgoing(self, reason):
+        """
+        Call the errback on all outstanding requests awaiting responses.
+
+        @param reason: the Failure instance to pass to those errbacks.
+        """
+        self._failAllReason = reason
+        OR = list(self._outstandingRequests.viewitems())
+        self._outstandingRequests = None  # we can never send another request
+        for key, value in OR:
+            value.errback(reason)
+
+    def _nextTag(self):
+        """
+        Generate protocol-local serial numbers for _ask keys.
+
+        @return: a string that has not yet been used on this connection.
+        """
+        self._counter += 1
+        return '%x' % (self._counter,)
+
+    def _sendBoxCommand(self, command, box, requiresAnswer=True):
+        """
+        Send a command across the wire with the given C{amp32.Box}.
+
+        Mutate the given box to give it any additional keys (_command, _ask)
+        required for the command and request/response machinery, then send it.
+
+        If requiresAnswer is True, returns a C{Deferred} which fires when a
+        response is received. The C{Deferred} is fired with an C{amp32.Box} on
+        success, or with an C{amp32.RemoteAmpError} if an error is received.
+
+        If the Deferred fails and the error is not handled by the caller of
+        this method, the failure will be logged and the connection dropped.
+
+        @param command: a bytes, the name of the command to issue.
+
+        @param box: an AmpBox with the arguments for the command.
+
+        @param requiresAnswer: a boolean. Defaults to True. If True, return a
+        Deferred which will fire when the other side responds to this command.
+        If False, return None and do not ask the other side for
+        acknowledgement.
+
+        @return: a Deferred which fires the AmpBox that holds the response to
+        this command, or None, as specified by requiresAnswer.
+
+        @raise ProtocolSwitched: if the protocol has been switched.
+        """
+        if self._failAllReason is not None:
+            return fail(self._failAllReason)
+        box[COMMAND] = command
+        tag = self._nextTag()
+        if requiresAnswer:
+            box[ASK] = tag
+        box._sendTo(self.boxSender)
+        if requiresAnswer:
+            result = self._outstandingRequests[tag] = Deferred()
+        else:
+            result = None
+        return result
+
+    def callRemoteString(self, command, requiresAnswer=True, **kw):
+        """
+        This is a low-level API, designed only for optimizing simple messages
+        for which the overhead of parsing is too great.
+
+        @param command: a bytes naming the command.
+
+        @param kw: arguments to the amp box.
+
+        @param requiresAnswer: a boolean. Defaults to True. If True, return a
+        Deferred which will fire when the other side responds to this command.
+        If False, return None and do not ask the other side for
+        acknowledgement.
+
+        @return: a Deferred which fires the AmpBox that holds the response to
+        this command, or None, as specified by requiresAnswer.
+        """
+        box = Box(kw)
+        return self._sendBoxCommand(command, box, requiresAnswer)
+
+    def callRemote(self, commandType, *a, **kw):
+        """
+        This is the primary high-level API for sending messages via AMP.
+        Invoke it with a command and appropriate arguments to send a message
+        to this connection's peer.
+
+        @param commandType: a subclass of Command.
+        @type commandType: L{type}
+
+        @param a: Positional (special) parameters taken by the command.
+        Positional parameters will typically not be sent over the wire.  The
+        only command included with AMP which uses positional parameters is
+        L{ProtocolSwitchCommand}, which takes the protocol that will be
+        switched to as its first argument.
+
+        @param kw: Keyword arguments taken by the command.  These are the
+        arguments declared in the command's 'arguments' attribute.  They will
+        be encoded and sent to the peer as arguments for the L{commandType}.
+
+        @return: If L{commandType} has a C{requiresAnswer} attribute set to
+        L{False}, then return L{None}.  Otherwise, return a L{Deferred} which
+        fires with a dictionary of objects representing the result of this
+        call.  Additionally, this L{Deferred} may fail with an exception
+        representing a connection failure, with L{UnknownRemoteError} if the
+        other end of the connection fails for an unknown reason, or with any
+        error specified as a key in L{commandType}'s C{errors} dictionary.
+        """
+
+        # XXX this takes command subclasses and not command objects on purpose.
+        # There's really no reason to have all this back-and-forth between
+        # command objects and the protocol, and the extra object being created
+        # (the Command instance) is pointless.  Command is kind of like
+        # Interface, and should be more like it.
+
+        # In other words, the fact that commandType is instantiated here is an
+        # implementation detail.  Don't rely on it.
+
+        try:
+            co = commandType(*a, **kw)
+        except:
+            return fail()
+        return co._doCommand(self)
+
+    def unhandledError(self, failure):
+        """
+        This is a terminal callback called after application code has had a
+        chance to quash any errors.
+        """
+        return self.boxSender.unhandledError(failure)
+
+    def _answerReceived(self, box):
+        """
+        An AMP box was received that answered a command previously sent with
+        L{callRemote}.
+
+        @param box: an AmpBox with a value for its L{ANSWER} key.
+        """
+        question = self._outstandingRequests.pop(box[ANSWER])
+        question.addErrback(self.unhandledError)
+        question.callback(box)
+
+    def _errorReceived(self, box):
+        """
+        An AMP box was received that answered a command previously sent with
+        L{callRemote}, with an error.
+
+        @param box: an L{AmpBox} with a value for its L{ERROR}, L{ERROR_CODE},
+        and L{ERROR_DESCRIPTION} keys.
+        """
+        question = self._outstandingRequests.pop(box[ERROR])
+        question.addErrback(self.unhandledError)
+        errorCode = box[ERROR_CODE]
+        description = box[ERROR_DESCRIPTION]
+        if errorCode in PROTOCOL_ERRORS:
+            exc = PROTOCOL_ERRORS[errorCode](errorCode, description)
+        else:
+            exc = RemoteAmpError(errorCode, description)
+        question.errback(Failure(exc))
+
+    def _commandReceived(self, box):
+        """
+        @param box: an L{AmpBox} with a value for its L{COMMAND} and L{ASK}
+        keys.
+        """
+        def formatAnswer(answerBox):
+            answerBox[ANSWER] = box[ASK]
+            return answerBox
+
+        def formatError(error):
+            if error.check(RemoteAmpError):
+                code = error.value.errorCode
+                desc = error.value.description
+                if error.value.fatal:
+                    errorBox = QuitBox()
+                else:
+                    errorBox = AmpBox()
+            else:
+                errorBox = QuitBox()
+                # here is where server-side logging happens if the error isn't
+                # handled
+                log.err(error)
+                code = UNKNOWN_ERROR_CODE
+                desc = "Unknown Error"
+            errorBox[ERROR] = box[ASK]
+            errorBox[ERROR_DESCRIPTION] = desc
+            errorBox[ERROR_CODE] = code
+            return errorBox
+
+        deferred = self.dispatchCommand(box)
+        if ASK in box:
+            deferred.addCallbacks(formatAnswer, formatError)
+            deferred.addCallback(self._safeEmit)
+        deferred.addErrback(self.unhandledError)
+
+    def ampBoxReceived(self, box):
+        """
+        An AmpBox was received, representing a command, or an answer to a
+        previously issued command (either successful or erroneous).  Respond to
+        it according to its contents.
+
+        @param box: an AmpBox
+
+        @raise NoEmptyBoxes: when a box is received that does not contain an
+        '_answer', '_command' / '_ask', or '_error' key; i.e. one which does
+        not fit into the command / response protocol defined by AMP.
+        """
+        if ANSWER in box:
+            self._answerReceived(box)
+        elif ERROR in box:
+            self._errorReceived(box)
+        elif COMMAND in box:
+            self._commandReceived(box)
+        else:
+            raise NoEmptyBoxes(box)
+
+    def _safeEmit(self, aBox):
+        """
+        Emit a box, ignoring L{ProtocolSwitched} and L{ConnectionLost} errors
+        which cannot be usefully handled.
+        """
+        try:
+            aBox._sendTo(self.boxSender)
+        except (ProtocolSwitched, ConnectionLost):
+            pass
+
+    def dispatchCommand(self, box):
+        """
+        A box with a _command key was received.
+
+        Dispatch it to a local handler call it.
+
+        @param proto: an AMP instance.
+        @param box: an AmpBox to be dispatched.
+        """
+        cmd = box[COMMAND]
+        responder = self.locator.locateResponder(cmd)
+        if responder is None:
+            return fail(RemoteAmpError(
+                UNHANDLED_ERROR_CODE, "Unhandled Command: %r" % (cmd,),
+                False, local=Failure(UnhandledCommand())))
+        return maybeDeferred(responder, box)
+
+
+class CommandLocator:
+    """
+    A L{CommandLocator} is a collection of responders to AMP L{Command}s, with
+    the help of the L{Command.responder} decorator.
+    """
+
+    class __metaclass__(type):
+        """
+        This metaclass keeps track of all of the Command.responder-decorated
+        methods defined since the last CommandLocator subclass was defined.  It
+        assumes (usually correctly, but unfortunately not necessarily so) that
+        those commands responders were all declared as methods of the class
+        being defined.  Note that this list can be incorrect if users use the
+        Command.responder decorator outside the context of a CommandLocator
+        class declaration.
+
+        Command responders defined on subclasses are given precedence over
+        those inherited from a base class.
+
+        The Command.responder decorator explicitly cooperates with this
+        metaclass.
+        """
+
+        _currentClassCommands = []
+
+        def __new__(cls, name, bases, attrs):
+            commands = cls._currentClassCommands[:]
+            cls._currentClassCommands[:] = []
+            cd = attrs['_commandDispatch'] = {}
+            subcls = type.__new__(cls, name, bases, attrs)
+            ancestors = list(subcls.__mro__[1:])
+            ancestors.reverse()
+            for ancestor in ancestors:
+                cd.update(getattr(ancestor, '_commandDispatch', {}))
+            for commandClass, responderFunc in commands:
+                cd[commandClass.commandName] = (commandClass, responderFunc)
+            if (bases and (
+                    subcls.lookupFunction != CommandLocator.lookupFunction)):
+                def locateResponder(self, name):
+                    warnings.warn(
+                        "Override locateResponder, not lookupFunction.",
+                        category=PendingDeprecationWarning,
+                        stacklevel=2)
+                    return self.lookupFunction(name)
+                subcls.locateResponder = locateResponder
+            return subcls
+
+    implements(IResponderLocator)
+
+    def _wrapWithSerialization(self, aCallable, command):
+        """
+        Wrap aCallable with its command's argument de-serialization
+        and result serialization logic.
+
+        @param aCallable: a callable with a 'command' attribute, designed to be
+        called with keyword arguments.
+
+        @param command: the command class whose serialization to use.
+
+        @return: a 1-arg callable which, when invoked with an AmpBox, will
+        deserialize the argument list and invoke appropriate user code for the
+        callable's command, returning a Deferred which fires with the result or
+        fails with an error.
+        """
+        def doit(box):
+            kw = command.parseArguments(box, self)
+
+            def checkKnownErrors(error):
+                key = error.trap(*command.allErrors)
+                code = command.allErrors[key]
+                desc = bytes(error.value)
+                return Failure(RemoteAmpError(
+                    code, desc, key in command.fatalErrors, local=error))
+
+            def makeResponseFor(objects):
+                try:
+                    return command.makeResponse(objects, self)
+                except:
+                    # let's helpfully log this.
+                    originalFailure = Failure()
+                    raise BadLocalReturn(
+                        "%r returned %r and %r could not serialize it" % (
+                            aCallable,
+                            objects,
+                            command),
+                        originalFailure)
+
+            return maybeDeferred(aCallable, **kw).addCallback(
+                makeResponseFor).addErrback(
+                checkKnownErrors)
+
+        return doit
+
+    def lookupFunction(self, name):
+        """
+        Deprecated synonym for L{locateResponder}
+        """
+        if self.__class__.lookupFunction != CommandLocator.lookupFunction:
+            return CommandLocator.locateResponder(self, name)
+        else:
+            warnings.warn("Call locateResponder, not lookupFunction.",
+                          category=PendingDeprecationWarning,
+                          stacklevel=2)
+        return self.locateResponder(name)
+
+    def locateResponder(self, name):
+        """
+        Locate a callable to invoke when executing the named command.
+
+        @param name: the normalized name (from the wire) of the command.
+
+        @return: a 1-argument function that takes a Box and returns a box or a
+        Deferred which fires a Box, for handling the command identified by the
+        given name, or None, if no appropriate responder can be found.
+        """
+        # Try to find a high-level method to invoke, and if we can't find one,
+        # fall back to a low-level one.
+        cd = self._commandDispatch
+        if name in cd:
+            commandClass, responderFunc = cd[name]
+            responderMethod = types.MethodType(
+                responderFunc, self, self.__class__)
+            return self._wrapWithSerialization(responderMethod, commandClass)
+
+
+class SimpleStringLocator(object):
+    """
+    Implement the L{locateResponder} method to do simple, string-based
+    dispatch.
+    """
+
+    implements(IResponderLocator)
+
+    baseDispatchPrefix = 'amp_'
+
+    def locateResponder(self, name):
+        """
+        Locate a callable to invoke when executing the named command.
+
+        @return: a function with the name C{"amp_" + name} on L{self}, or None
+        if no such function exists.  This function will then be called with the
+        L{AmpBox} itself as an argument.
+
+        @param name: the normalized name (from the wire) of the command.
+        """
+        fName = self.baseDispatchPrefix + (name.upper())
+        return getattr(self, fName, None)
+
+
+PYTHON_KEYWORDS = [
+    'and', 'del', 'for', 'is', 'raise', 'assert', 'elif', 'from', 'lambda',
+    'return', 'break', 'else', 'global', 'not', 'try', 'class', 'except',
+    'if', 'or', 'while', 'continue', 'exec', 'import', 'pass', 'yield',
+    'def', 'finally', 'in', 'print']
+
+
+def _wireNameToPythonIdentifier(key):
+    """
+    (Private) Normalize an argument name from the wire for use with Python
+    code.  If the return value is going to be a python keyword it will be
+    capitalized.  If it contains any dashes they will be replaced with
+    underscores.
+
+    The rationale behind this method is that AMP should be an inherently
+    multi-language protocol, so message keys may contain all manner of bizarre
+    bytes.  This is not a complete solution; there are still forms of arguments
+    that this implementation will be unable to parse.  However, Python
+    identifiers share a huge raft of properties with identifiers from many
+    other languages, so this is a 'good enough' effort for now.  We deal
+    explicitly with dashes because that is the most likely departure: Lisps
+    commonly use dashes to separate method names, so protocols initially
+    implemented in a lisp amp dialect may use dashes in argument or command
+    names.
+
+    @param key: a bytes, looking something like 'foo-bar-baz' or 'from'
+
+    @return: a bytes which is a valid python identifier, looking something like
+    'foo_bar_baz' or 'From'.
+    """
+    lkey = key.replace("-", "_")
+    if lkey in PYTHON_KEYWORDS:
+        return lkey.title()
+    return lkey
+
+
+class Argument:
+    """
+    Base-class of all objects that take values from Amp packets and convert
+    them into objects for Python functions.
+
+    This implementation of L{IArgumentType} provides several higher-level
+    hooks for subclasses to override.  See L{toString} and L{fromString}
+    which will be used to define the behavior of L{IArgumentType.toBox} and
+    L{IArgumentType.fromBox}, respectively.
+    """
+
+    implements(IArgumentType)
+
+    optional = False
+
+    def __init__(self, optional=False):
+        """
+        Create an Argument.
+
+        @param optional: a boolean indicating whether this argument can be
+        omitted in the protocol.
+        """
+        self.optional = optional
+
+    def retrieve(self, d, name, proto):
+        """
+        Retrieve the given key from the given dictionary, removing it if found.
+
+        @param d: a dictionary.
+
+        @param name: a key in L{d}.
+
+        @param proto: an instance of an AMP.
+
+        @raise KeyError: if I am not optional and no value was found.
+
+        @return: d[name].
+        """
+        if self.optional:
+            value = d.get(name)
+            if value is not None:
+                del d[name]
+        else:
+            value = d.pop(name)
+        return value
+
+    def fromBox(self, name, strings, objects, proto):
+        """
+        Populate an 'out' dictionary with mapping names to Python values
+        decoded from an 'in' AmpBox mapping strings to string values.
+
+        @param name: the argument name to retrieve
+        @type name: bytes
+
+        @param strings: The AmpBox to read string(s) from, a mapping of
+        argument names to string values.
+        @type strings: AmpBox
+
+        @param objects: The dictionary to write object(s) to, a mapping of
+        names to Python objects.
+        @type objects: dict
+
+        @param proto: an AMP instance.
+        """
+        st = self.retrieve(strings, name, proto)
+        nk = _wireNameToPythonIdentifier(name)
+        if self.optional and st is None:
+            objects[nk] = None
+        else:
+            objects[nk] = self.fromStringProto(st, proto)
+
+    def toBox(self, name, strings, objects, proto):
+        """
+        Populate an 'out' AmpBox with strings encoded from an 'in' dictionary
+        mapping names to Python values.
+
+        @param name: the argument name to retrieve
+        @type name: bytes
+
+        @param strings: The AmpBox to write string(s) to, a mapping of
+        argument names to string values.
+        @type strings: AmpBox
+
+        @param objects: The dictionary to read object(s) from, a mapping of
+        names to Python objects.
+
+        @type objects: dict
+
+        @param proto: the protocol we are converting for.
+        @type proto: AMP
+        """
+        obj = self.retrieve(objects, _wireNameToPythonIdentifier(name), proto)
+        if self.optional and obj is None:
+            # strings[name] = None
+            pass
+        else:
+            strings[name] = self.toStringProto(obj, proto)
+
+    def fromStringProto(self, inString, proto):
+        """
+        Convert a string to a Python value.
+
+        @param inString: the string to convert.
+
+        @param proto: the protocol we are converting for.
+        @type proto: AMP
+
+        @return: a Python object.
+        """
+        return self.fromString(inString)
+
+    def toStringProto(self, inObject, proto):
+        """
+        Convert a Python object to a string.
+
+        @param inObject: the object to convert.
+
+        @param proto: the protocol we are converting for.
+        @type proto: AMP
+        """
+        return self.toString(inObject)
+
+    def fromString(self, inString):
+        """
+        Convert a string to a Python object.  Subclasses must implement this.
+
+        @param inString: the string to convert.
+        @type inString: bytes
+
+        @return: the decoded value from inString
+        """
+
+    def toString(self, inObject):
+        """
+        Convert a Python object into a string for passing over the network.
+
+        @param inObject: an object of the type that this Argument is intended
+        to deal with.
+
+        @return: the wire encoding of inObject
+        @rtype: bytes
+        """
+
+
+class Integer(Argument):
+    """
+    Encode any integer values of any size on the wire as the string
+    representation.
+
+    Example: C{123} becomes C{"123"}
+    """
+
+    fromString = int
+
+    def toString(self, inObject):
+        return bytes(int(inObject))
+
+
+class String(Argument):
+    """
+    Don't do any conversion at all; just pass through 'bytes'.
+    """
+
+    def toString(self, inObject):
+        return inObject
+
+    def fromString(self, inString):
+        return inString
+
+
+class Float(Argument):
+    """
+    Encode floating-point values on the wire as their repr.
+    """
+
+    fromString = float
+    toString = repr
+
+
+class Boolean(Argument):
+    """
+    Encode True or False as "True" or "False" on the wire.
+    """
+
+    def fromString(self, inString):
+        if inString == 'True':
+            return True
+        elif inString == 'False':
+            return False
+        else:
+            raise TypeError("Bad boolean value: %r" % (inString,))
+
+    def toString(self, inObject):
+        if inObject:
+            return 'True'
+        else:
+            return 'False'
+
+
+class Unicode(String):
+    """
+    Encode a unicode string on the wire as UTF-8.
+    """
+
+    def toString(self, inObject):
+        # assert isinstance(inObject, unicode)
+        return String.toString(self, inObject.encode('utf-8'))
+
+    def fromString(self, inString):
+        # assert isinstance(inString, bytes)
+        return String.fromString(self, inString).decode('utf-8')
+
+
+class Path(Unicode):
+    """
+    Encode and decode L{filepath.FilePath} instances as paths on the wire.
+
+    This is really intended for use with subprocess communication tools:
+    exchanging pathnames on different machines over a network is not generally
+    meaningful, but neither is it disallowed; you can use this to communicate
+    about NFS paths, for example.
+    """
+
+    def fromString(self, inString):
+        return filepath.FilePath(Unicode.fromString(self, inString))
+
+    def toString(self, inObject):
+        return Unicode.toString(self, inObject.path)
+
+
+class ListOf(Argument):
+    """
+    Encode and decode lists of instances of a single other argument type.
+
+    For example, if you want to pass::
+
+        [3, 7, 9, 15]
+
+    You can create an argument like this::
+
+        ListOf(Integer())
+
+    The serialized form of the entire list is subject to the limit imposed by
+    L{MAX_VALUE_LENGTH}.  List elements are represented as 32-bit length
+    prefixed strings.  The argument type passed to the L{ListOf} initializer is
+    responsible for producing the serialized form of each element.
+
+    @ivar elementType: The L{Argument} instance used to encode and decode list
+        elements (note, not an arbitrary L{IArgument} implementation:
+        arguments must be implemented using only the C{fromString} and
+        C{toString} methods, not the C{fromBox} and C{toBox} methods).
+
+    @param optional: a boolean indicating whether this argument can be
+        omitted in the protocol.
+
+    @since: 10.0
+    """
+
+    def __init__(self, elementType, optional=False):
+        self.elementType = elementType
+        Argument.__init__(self, optional)
+
+    def fromString(self, inString):
+        """
+        Convert the serialized form of a list of instances of some type back
+        into that list.
+        """
+        strings = []
+        parser = Int32StringReceiver()
+        parser.MAX_LENGTH = MAX_VALUE_LENGTH
+        parser.stringReceived = strings.append
+        parser.dataReceived(inString)
+        return map(self.elementType.fromString, strings)
+
+    def toString(self, inObject):
+        """
+        Serialize the given list of objects to a single string.
+        """
+        strings = []
+        for obj in inObject:
+            serialized = self.elementType.toString(obj)
+            strings.append(pack('!I', len(serialized)))
+            strings.append(serialized)
+        return ''.join(strings)
+
+
+class AmpList(Argument):
+    """
+    Convert a list of dictionaries into a list of AMP boxes on the wire.
+
+    For example, if you want to pass::
+
+        [{'a': 7, 'b': u'hello'}, {'a': 9, 'b': u'goodbye'}]
+
+    You might use an AmpList like this in your arguments or response list::
+
+        AmpList([('a', Integer()),
+                 ('b', Unicode())])
+    """
+
+    def __init__(self, subargs, optional=False):
+        """
+        Create an AmpList.
+
+        @param subargs: a list of 2-tuples of ('name', argument) describing the
+        schema of the dictionaries in the sequence of amp boxes.
+
+        @param optional: a boolean indicating whether this argument can be
+        omitted in the protocol.
+        """
+        self.subargs = subargs
+        Argument.__init__(self, optional)
+
+    def fromStringProto(self, inString, proto):
+        boxes = parseString(inString)
+        values = [_stringsToObjects(box, self.subargs, proto)
+                  for box in boxes]
+        return values
+
+    def toStringProto(self, inObject, proto):
+        return ''.join(
+            _objectsToStrings(objects, self.subargs, Box(), proto).serialize()
+            for objects in inObject)
+
+
+class Descriptor(Integer):
+    """
+    Encode and decode file descriptors for exchange over a UNIX domain socket.
+
+    This argument type requires an AMP connection set up over an
+    L{IUNIXTransport<twisted.internet.interfaces.IUNIXTransport>} provider (for
+    example, the kind of connection created by
+    L{IReactorUNIX.connectUNIX<
+        twisted.internet.interfaces.IReactorUNIX.connectUNIX>}
+    and L{UNIXClientEndpoint<twisted.internet.endpoints.UNIXClientEndpoint>}).
+
+    There is no correspondence between the integer value of the file
+    descriptor on the sending and receiving sides, therefore an alternate
+    approach is taken to matching up received descriptors with particular
+    L{Descriptor} parameters. The argument is encoded to an ordinal (unique
+    per connection) for inclusion in the AMP command or response box. The
+    descriptor itself is sent using
+    L{IUNIXTransport.sendFileDescriptor<
+        twisted.internet.interfaces.IUNIXTransport.sendFileDescriptor>}.
+    The receiver uses the order in which file descriptors are received and the
+    ordinal value to come up with the received copy of the descriptor.
+    """
+
+    def fromStringProto(self, inString, proto):
+        """
+        Take a unique identifier associated with a file descriptor which must
+        have been received by now and use it to look up that descriptor in a
+        dictionary where they are kept.
+
+        @param inString: The base representation (as a byte string) of an
+            ordinal indicating which file descriptor corresponds to this usage
+            of this argument.
+        @type inString: C{bytes}
+
+        @param proto: The protocol used to receive this descriptor.  This
+            protocol must be connected via a transport providing
+            L{IUNIXTransport<twisted.internet.interfaces.IUNIXTransport>}.
+        @type proto: L{BinaryBoxProtocol}
+
+        @return: The file descriptor represented by C{inString}.
+        @rtype: C{int}
+        """
+        return proto._getDescriptor(int(inString))
+
+    def toStringProto(self, inObject, proto):
+        """
+        Send C{inObject}, an integer file descriptor, over C{proto}'s
+        connection and return a unique identifier which will allow the
+        receiver to associate the file descriptor with this argument.
+
+        @param inObject: A file descriptor to duplicate over an AMP connection
+            as the value for this argument.
+        @type inObject: C{int}
+
+        @param proto: The protocol which will be used to send this descriptor.
+            This protocol must be connected via a transport providing
+            L{IUNIXTransport<twisted.internet.interfaces.IUNIXTransport>}.
+
+        @return: A byte string which can be used by the receiver to reconstruct
+            the file descriptor.
+        @type: C{bytes}
+        """
+        identifier = proto._sendFileDescriptor(inObject)
+        outString = Integer.toStringProto(self, identifier, proto)
+        return outString
+
+
+class Command:
+    """
+    Subclass me to specify an AMP Command.
+
+    @cvar arguments: A list of 2-tuples of (name, Argument-subclass-instance),
+    specifying the names and values of the parameters which are required for
+    this command.
+
+    @cvar response: A list like L{arguments}, but instead used for the return
+    value.
+
+    @cvar errors: A mapping of subclasses of L{Exception} to wire-protocol tags
+    for errors represented as L{bytes}s.  Responders which raise keys from this
+    dictionary will have the error translated to the corresponding tag on the
+    wire.  Invokers which receive Deferreds from invoking this command with
+    L{AMP.callRemote} will potentially receive Failures with keys from this
+    mapping as their value.  This mapping is inherited; if you declare a
+    command which handles C{FooError} as 'FOO_ERROR', then subclass it and
+    specify C{BarError} as 'BAR_ERROR', responders to the subclass may raise
+    either C{FooError} or C{BarError}, and invokers must be able to deal with
+    either of those exceptions.
+
+    @cvar fatalErrors: like 'errors', but errors in this list will always
+    terminate the connection, despite being of a recognizable error type.
+
+    @cvar commandType: The type of Box used to issue commands; useful only for
+    protocol-modifying behavior like startTLS or protocol switching.  Defaults
+    to a plain vanilla L{Box}.
+
+    @cvar responseType: The type of Box used to respond to this command; only
+    useful for protocol-modifying behavior like startTLS or protocol switching.
+    Defaults to a plain vanilla L{Box}.
+
+    @ivar requiresAnswer: a boolean; defaults to True.  Set it to False on your
+    subclass if you want callRemote to return None.  Note: this is a hint only
+    to the client side of the protocol.  The return-type of a command responder
+    method must always be a dictionary adhering to the contract specified by
+    L{response}, because clients are always free to request a response if they
+    want one.
+    """
+
+    class __metaclass__(type):
+        """
+        Metaclass hack to establish reverse-mappings for 'errors' and
+        'fatalErrors' as class vars.
+        """
+        def __new__(cls, name, bases, attrs):
+            reverseErrors = attrs['reverseErrors'] = {}
+            er = attrs['allErrors'] = {}
+            if 'commandName' not in attrs:
+                attrs['commandName'] = name
+            newtype = type.__new__(cls, name, bases, attrs)
+            errors = {}
+            fatalErrors = {}
+            accumulateClassDict(newtype, 'errors', errors)
+            accumulateClassDict(newtype, 'fatalErrors', fatalErrors)
+            for v, k in errors.viewitems():
+                reverseErrors[k] = v
+                er[v] = k
+            for v, k in fatalErrors.viewitems():
+                reverseErrors[k] = v
+                er[v] = k
+            return newtype
+
+    arguments = []
+    response = []
+    extra = []
+    errors = {}
+    fatalErrors = {}
+
+    commandType = Box
+    responseType = Box
+
+    requiresAnswer = True
+
+    def __init__(self, **kw):
+        """
+        Create an instance of this command with specified values for its
+        parameters.
+
+        @param kw: a dict containing an appropriate value for each name
+        specified in the L{arguments} attribute of my class.
+
+        @raise InvalidSignature: if you forgot any required arguments.
+        """
+        self.structured = kw
+        givenArgs = set(kw)
+        forgotten = []
+        for name, arg in self.arguments:
+            pythonName = _wireNameToPythonIdentifier(name)
+            if pythonName not in givenArgs and not arg.optional:
+                forgotten.append(pythonName)
+        if forgotten:
+            raise InvalidSignature("forgot %s for %s" % (
+                ', '.join(forgotten), self.commandName))
+        forgotten = []
+
+    @classmethod
+    def makeResponse(cls, objects, proto):
+        """
+        Serialize a mapping of arguments using this L{Command}'s
+        response schema.
+
+        @param objects: a dict with keys matching the names specified in
+        self.response, having values of the types that the Argument objects in
+        self.response can format.
+
+        @param proto: an L{AMP}.
+
+        @return: an L{AmpBox}.
+        """
+        try:
+            responseType = cls.responseType()
+        except:
+            return fail()
+        return _objectsToStrings(objects, cls.response, responseType, proto)
+
+    @classmethod
+    def makeArguments(cls, objects, proto):
+        """
+        Serialize a mapping of arguments using this L{Command}'s
+        argument schema.
+
+        @param objects: a dict with keys similar to the names specified in
+        self.arguments, having values of the types that the Argument objects in
+        self.arguments can parse.
+
+        @param proto: an L{AMP}.
+
+        @return: An instance of this L{Command}'s C{commandType}.
+        """
+        allowedNames = set()
+        for (argName, ignored) in cls.arguments:
+            allowedNames.add(_wireNameToPythonIdentifier(argName))
+
+        for intendedArg in objects:
+            if intendedArg not in allowedNames:
+                raise InvalidSignature(
+                    "%s is not a valid argument" % (intendedArg,))
+        return _objectsToStrings(objects, cls.arguments, cls.commandType(),
+                                 proto)
+
+    @classmethod
+    def parseResponse(cls, box, protocol):
+        """
+        Parse a mapping of serialized arguments using this
+        L{Command}'s response schema.
+
+        @param box: A mapping of response-argument names to the
+        serialized forms of those arguments.
+        @param protocol: The L{AMP} protocol.
+
+        @return: A mapping of response-argument names to the parsed
+        forms.
+        """
+        return _stringsToObjects(box, cls.response, protocol)
+
+    @classmethod
+    def parseArguments(cls, box, protocol):
+        """
+        Parse a mapping of serialized arguments using this
+        L{Command}'s argument schema.
+
+        @param box: A mapping of argument names to the seralized forms
+        of those arguments.
+        @param protocol: The L{AMP} protocol.
+
+        @return: A mapping of argument names to the parsed forms.
+        """
+        return _stringsToObjects(box, cls.arguments, protocol)
+
+    @classmethod
+    def responder(cls, methodfunc):
+        """
+        Declare a method to be a responder for a particular command.
+
+        This is a decorator.
+
+        Use like so::
+
+            class MyCommand(Command):
+                arguments = [('a', ...), ('b', ...)]
+
+            class MyProto(AMP):
+                def myFunMethod(self, a, b):
+                    ...
+                MyCommand.responder(myFunMethod)
+
+        Notes: Although decorator syntax is not used within Twisted, this
+        function returns its argument and is therefore safe to use with
+        decorator syntax.
+
+        This is not thread safe.  Don't declare AMP subclasses in other
+        threads.  Don't declare responders outside the scope of AMP subclasses;
+        the behavior is undefined.
+
+        @param methodfunc: A function which will later become a method, which
+        has a keyword signature compatible with this command's L{argument} list
+        and returns a dictionary with a set of keys compatible with this
+        command's L{response} list.
+
+        @return: the methodfunc parameter.
+        """
+        CommandLocator._currentClassCommands.append((cls, methodfunc))
+        return methodfunc
+
+    # Our only instance method
+    def _doCommand(self, proto):
+        """
+        Encode and send this Command to the given protocol.
+
+        @param proto: an AMP, representing the connection to send to.
+
+        @return: a Deferred which will fire or error appropriately when the
+        other side responds to the command (or error if the connection is lost
+        before it is responded to).
+        """
+
+        def _massageError(error):
+            error.trap(RemoteAmpError)
+            rje = error.value
+            errorType = self.reverseErrors.get(rje.errorCode,
+                                               UnknownRemoteError)
+            return Failure(errorType(rje.description))
+
+        d = proto._sendBoxCommand(self.commandName,
+                                  self.makeArguments(self.structured, proto),
+                                  self.requiresAnswer)
+
+        if self.requiresAnswer:
+            d.addCallback(self.parseResponse, proto)
+            d.addErrback(_massageError)
+
+        return d
+
+
+class _NoCertificate:
+    """
+    This is for peers which don't want to use a local certificate.  Used by
+    AMP because AMP's internal language is all about certificates and this
+    duck-types in the appropriate place; this API isn't really stable though,
+    so it's not exposed anywhere public.
+
+    For clients, it will use ephemeral DH keys, or whatever the default is for
+    certificate-less clients in OpenSSL.  For servers, it will generate a
+    temporary self-signed certificate with garbage values in the DN and use
+    that.
+    """
+
+    def __init__(self, client):
+        """
+        Create a _NoCertificate which either is or isn't for the client side of
+        the connection.
+
+        @param client: True if we are a client and should truly have no
+        certificate and be anonymous, False if we are a server and actually
+        have to generate a temporary certificate.
+
+        @type client: bool
+        """
+        self.client = client
+
+    def options(self, *authorities):
+        """
+        Behaves like L{twisted.internet.ssl.PrivateCertificate.options}().
+        """
+        if not self.client:
+            # do some crud with sslverify to generate a temporary self-signed
+            # certificate.  This is SLOOOWWWWW so it is only in the absolute
+            # worst, most naive case.
+
+            # We have to do this because OpenSSL will not let both the server
+            # and client be anonymous.
+            sharedDN = DN(CN='TEMPORARY CERTIFICATE')
+            key = KeyPair.generate()
+            cr = key.certificateRequest(sharedDN)
+            sscrd = key.signCertificateRequest(
+                sharedDN, cr, lambda dn: True, 1)
+            cert = key.newCertificate(sscrd)
+            return cert.options(*authorities)
+        options = dict()
+        if authorities:
+            options.update(
+                verify=True, requireCertificate=True,
+                caCerts=[auth.original for auth in authorities])
+        occo = CertificateOptions(**options)
+        return occo
+
+
+class _TLSBox(AmpBox):
+    """
+    I am an AmpBox that, upon being sent, initiates a TLS connection.
+    """
+
+    __slots__ = []
+
+    def __init__(self):
+        if ssl is None:
+            raise RemoteAmpError("TLS_ERROR", "TLS not available")
+        AmpBox.__init__(self)
+
+    def _keyprop(k, default):
+        return property(lambda self: self.get(k, default))
+
+    # These properties are described in startTLS
+    certificate = _keyprop('tls_localCertificate', _NoCertificate(False))
+    verify = _keyprop('tls_verifyAuthorities', None)
+
+    def _sendTo(self, proto):
+        """
+        Send my encoded value to the protocol, then initiate TLS.
+        """
+        ab = AmpBox(self)
+        for k in ['tls_localCertificate',
+                  'tls_verifyAuthorities']:
+            ab.pop(k, None)
+        ab._sendTo(proto)
+        proto._startTLS(self.certificate, self.verify)
+
+
+class _LocalArgument(String):
+    """
+    Local arguments are never actually relayed across the wire.  This is just a
+    shim so that StartTLS can pretend to have some arguments: if arguments
+    acquire documentation properties, replace this with something nicer later.
+    """
+
+    def fromBox(self, name, strings, objects, proto):
+        pass
+
+
+class StartTLS(Command):
+    """
+    Use, or subclass, me to implement a command that starts TLS.
+
+    Callers of StartTLS may pass several special arguments, which affect the
+    TLS negotiation:
+
+        - tls_localCertificate: This is a
+        twisted.internet.ssl.PrivateCertificate which will be used to secure
+        the side of the connection it is returned on.
+
+        - tls_verifyAuthorities: This is a list of
+        twisted.internet.ssl.Certificate objects that will be used as the
+        certificate authorities to verify our peer's certificate.
+
+    Each of those special parameters may also be present as a key in the
+    response dictionary.
+    """
+
+    arguments = [("tls_localCertificate", _LocalArgument(optional=True)),
+                 ("tls_verifyAuthorities", _LocalArgument(optional=True))]
+
+    response = [("tls_localCertificate", _LocalArgument(optional=True)),
+                ("tls_verifyAuthorities", _LocalArgument(optional=True))]
+
+    responseType = _TLSBox
+
+    def __init__(self, **kw):
+        """
+        Create a StartTLS command.  (This is private.  Use AMP.callRemote.)
+
+        @param tls_localCertificate: the PrivateCertificate object to use to
+        secure the connection.  If it's None, or unspecified, an ephemeral DH
+        key is used instead.
+
+        @param tls_verifyAuthorities: a list of Certificate objects which
+        represent root certificates to verify our peer with.
+        """
+        if ssl is None:
+            raise RuntimeError("TLS not available.")
+        self.certificate = kw.pop('tls_localCertificate', _NoCertificate(True))
+        self.authorities = kw.pop('tls_verifyAuthorities', None)
+        Command.__init__(self, **kw)
+
+    def _doCommand(self, proto):
+        """
+        When a StartTLS command is sent, prepare to start TLS, but don't
+        actually do it; wait for the acknowledgement, then initiate the TLS
+        handshake.
+        """
+        d = Command._doCommand(self, proto)
+        proto._prepareTLS(self.certificate, self.authorities)
+
+        # XXX before we get back to user code we are going to start TLS...
+        def actuallystart(response):
+            proto._startTLS(self.certificate, self.authorities)
+            return response
+        d.addCallback(actuallystart)
+
+        return d
+
+
+class ProtocolSwitchCommand(Command):
+    """
+    Use this command to switch from something Amp-derived to a different
+    protocol mid-connection.  This can be useful to use amp as the
+    connection-startup negotiation phase.  Since TLS is a different layer
+    entirely, you can use Amp to negotiate the security parameters of your
+    connection, then switch to a different protocol, and the connection will
+    remain secured.
+    """
+
+    def __init__(self, _protoToSwitchToFactory, **kw):
+        """
+        Create a ProtocolSwitchCommand.
+
+        @param _protoToSwitchToFactory: a ProtocolFactory which will generate
+        the Protocol to switch to.
+
+        @param kw: Keyword arguments, encoded and handled normally as
+        L{Command} would.
+        """
+
+        self.protoToSwitchToFactory = _protoToSwitchToFactory
+        super(ProtocolSwitchCommand, self).__init__(**kw)
+
+    @classmethod
+    def makeResponse(cls, innerProto, proto):
+        return _SwitchBox(innerProto)
+
+    def _doCommand(self, proto):
+        """
+        When we emit a ProtocolSwitchCommand, lock the protocol, but don't
+        actually switch to the new protocol unless an acknowledgement is
+        received. If an error is received, switch back.
+        """
+        d = super(ProtocolSwitchCommand, self)._doCommand(proto)
+        proto._lockForSwitch()
+
+        def switchNow(ign):
+            innerProto = self.protoToSwitchToFactory.buildProtocol(
+                proto.transport.getPeer())
+            proto._switchTo(innerProto, self.protoToSwitchToFactory)
+            return ign
+
+        def handle(ign):
+            proto._unlockFromSwitch()
+            self.protoToSwitchToFactory.clientConnectionFailed(
+                None, Failure(CONNECTION_LOST))
+            return ign
+
+        return d.addCallbacks(switchNow, handle)
+
+
+class _DescriptorExchanger(object):
+    """
+    L{_DescriptorExchanger} is a mixin for L{BinaryBoxProtocol} which adds
+    support for receiving file descriptors, a feature offered by
+    L{IUNIXTransport<twisted.internet.interfaces.IUNIXTransport>}.
+
+    @ivar _descriptors: Temporary storage for all file descriptors received.
+        Values in this dictionary are the file descriptors (as integers).  Keys
+        in this dictionary are ordinals giving the order in which each
+        descriptor was received.  The ordering information is used to allow
+        L{Descriptor} to determine which is the correct descriptor for any
+        particular usage of that argument type.
+    @type _descriptors: C{dict}
+
+    @ivar _sendingDescriptorCounter: A no-argument callable which returns the
+        ordinals, starting from 0.  This is used to construct values for
+        C{_sendFileDescriptor}.
+
+    @ivar _receivingDescriptorCounter: A no-argument callable which returns the
+        ordinals, starting from 0.  This is used to construct values for
+        C{fileDescriptorReceived}.
+    """
+
+    implements(IFileDescriptorReceiver)
+
+    def __init__(self):
+        self._descriptors = {}
+        self._getDescriptor = self._descriptors.pop
+        self._sendingDescriptorCounter = count()
+        self._receivingDescriptorCounter = count()
+
+    def _sendFileDescriptor(self, descriptor):
+        """
+        Assign and return the next ordinal to the given descriptor after
+        sending the descriptor over this protocol's transport.
+        """
+        self.transport.sendFileDescriptor(descriptor)
+        return next(self._sendingDescriptorCounter)
+
+    def fileDescriptorReceived(self, descriptor):
+        """
+        Collect received file descriptors to be claimed later by L{Descriptor}.
+
+        @param descriptor: The received file descriptor.
+        @type descriptor: C{int}
+        """
+        self._descriptors[next(self._receivingDescriptorCounter)] = descriptor
+
+
+class BinaryBoxProtocol(StatefulStringProtocol, Int32StringReceiver,
+                        _DescriptorExchanger):
+    """
+    A protocol for receiving L{AmpBox}es - key/value pairs - via
+    length-prefixed strings. A box is composed of:
+
+        - any number of key-value pairs, described by:
+            - a 4-byte network-endian packed key length (of which the first
+              3 bytes must be null, and the last must be non-null: i.e. the
+              value of the length must be 1-255)
+            - a key, comprised of that many bytes
+            - a 4-byte network-endian unsigned value length (up to the maximum
+              of C{MAX_VALUE_LENGTH})
+            - a value, comprised of that many bytes
+        - 4 null bytes
+
+    In other words, an even number of strings prefixed with packed unsigned
+    32-bit integers, and then a 0-length string to indicate the end of the box.
+
+    This protocol also implements 2 extra private bits of functionality related
+    to the byte boundaries between messages; it can start TLS between two given
+    boxes or switch to an entirely different protocol.  However, due to some
+    tricky elements of the implementation, the public interface to this
+    functionality is L{ProtocolSwitchCommand} and L{StartTLS}.
+
+    @ivar _keyLengthLimitExceeded: A flag which is only true when the
+        connection is being closed because a key length prefix which was longer
+        than allowed by the protocol was received.
+
+    @ivar boxReceiver: an L{IBoxReceiver} provider, whose L{ampBoxReceived}
+    method will be invoked for each L{AmpBox} that is received.
+    """
+
+    implements(IBoxSender)
+
+    _justStartedTLS = False
+    _startingTLSBuffer = None
+    _locked = False
+    _currentKey = None
+    _currentBox = None
+
+    _keyLengthLimitExceeded = False
+
+    hostCertificate = None
+    noPeerCertificate = False   # for tests
+    innerProtocol = None
+    innerProtocolClientFactory = None
+
+    def __init__(self, boxReceiver):
+        _DescriptorExchanger.__init__(self)
+        self.boxReceiver = boxReceiver
+
+    def _switchTo(self, newProto, clientFactory=None):
+        """
+        Switch this BinaryBoxProtocol's transport to a new protocol.  You need
+        to do this 'simultaneously' on both ends of a connection; the easiest
+        way to do this is to use a subclass of ProtocolSwitchCommand.
+
+        @param newProto: the new protocol instance to switch to.
+
+        @param clientFactory: the ClientFactory to send the
+        L{clientConnectionLost} notification to.
+        """
+        # All the data that Int32StringReceiver has not yet dealt with belongs
+        # to our new protocol: luckily it's keeping that in a handy (although
+        # ostensibly internal) variable for us:
+        newProtoData = self.recvd
+        # We're quite possibly in the middle of a 'dataReceived' loop in
+        # Int32StringReceiver: let's make sure that the next iteration, the
+        # loop will break and not attempt to look at something that isn't a
+        # length prefix.
+        self.recvd = ''
+        # Finally, do the actual work of setting up the protocol and delivering
+        # its first chunk of data, if one is available.
+        self.innerProtocol = newProto
+        self.innerProtocolClientFactory = clientFactory
+        newProto.makeConnection(self.transport)
+        if newProtoData:
+            newProto.dataReceived(newProtoData)
+
+    def sendBox(self, box):
+        """
+        Send a amp32.Box to my peer.
+
+        Note: transport.write is never called outside of this method.
+
+        @param box: an AmpBox.
+
+        @raise ProtocolSwitched: if the protocol has previously been switched.
+
+        @raise ConnectionLost: if the connection has previously been lost.
+        """
+        if self._locked:
+            raise ProtocolSwitched(
+                "This connection has switched: no AMP traffic allowed.")
+        if self.transport is None:
+            raise ConnectionLost()
+        if self._startingTLSBuffer is not None:
+            self._startingTLSBuffer.append(box)
+        else:
+            self.transport.write(box.serialize())
+
+    def makeConnection(self, transport):
+        """
+        Notify L{boxReceiver} that it is about to receive boxes from this
+        protocol by invoking L{startReceivingBoxes}.
+        """
+        self.transport = transport
+        self.boxReceiver.startReceivingBoxes(self)
+        self.connectionMade()
+
+    def dataReceived(self, data):
+        """
+        Either parse incoming data as L{AmpBox}es or relay it to our nested
+        protocol.
+        """
+        if self._justStartedTLS:
+            self._justStartedTLS = False
+        # If we already have an inner protocol, then we don't deliver data to
+        # the protocol parser any more; we just hand it off.
+        if self.innerProtocol is not None:
+            self.innerProtocol.dataReceived(data)
+            return
+        return Int32StringReceiver.dataReceived(self, data)
+
+    def connectionLost(self, reason):
+        """
+        The connection was lost; notify any nested protocol.
+        """
+        if self.innerProtocol is not None:
+            self.innerProtocol.connectionLost(reason)
+            if self.innerProtocolClientFactory is not None:
+                self.innerProtocolClientFactory.clientConnectionLost(
+                    None, reason)
+        if self._keyLengthLimitExceeded:
+            failReason = Failure(TooLong(True, False, None, None))
+        elif reason.check(ConnectionClosed) and self._justStartedTLS:
+            # We just started TLS and haven't received any data.  This means
+            # the other connection didn't like our cert (although they may not
+            # have told us why - later Twisted should make 'reason' into a TLS
+            # error.)
+            failReason = PeerVerifyError(
+                "Peer rejected our certificate for an unknown reason.")
+        else:
+            failReason = reason
+        self.boxReceiver.stopReceivingBoxes(failReason)
+
+    # The longest key allowed
+    _MAX_KEY_LENGTH = MAX_KEY_LENGTH
+
+    # The longest value allowed (this is somewhat redundant, as longer values
+    # cannot be encoded - ah well).
+    _MAX_VALUE_LENGTH = MAX_VALUE_LENGTH
+
+    # The first thing received is a key.
+    MAX_LENGTH = _MAX_KEY_LENGTH
+
+    def proto_init(self, string):
+        """
+        String received in the 'init' state.
+        """
+        self._currentBox = AmpBox()
+        return self.proto_key(string)
+
+    def proto_key(self, string):
+        """
+        String received in the 'key' state.  If the key is empty, a complete
+        box has been received.
+        """
+        if string:
+            self._currentKey = string
+            self.MAX_LENGTH = self._MAX_VALUE_LENGTH
+            return 'value'
+        else:
+            self.boxReceiver.ampBoxReceived(self._currentBox)
+            self._currentBox = None
+            return 'init'
+
+    def proto_value(self, string):
+        """
+        String received in the 'value' state.
+        """
+        self._currentBox[self._currentKey] = string
+        self._currentKey = None
+        self.MAX_LENGTH = self._MAX_KEY_LENGTH
+        return 'key'
+
+    def lengthLimitExceeded(self, length):
+        """
+        The key length limit was exceeded.  Disconnect the transport and make
+        sure a meaningful exception is reported.
+        """
+        self._keyLengthLimitExceeded = True
+        self.transport.loseConnection()
+
+    def _lockForSwitch(self):
+        """
+        Lock this binary protocol so that no further boxes may be sent.  This
+        is used when sending a request to switch underlying protocols.  You
+        probably want to subclass ProtocolSwitchCommand rather than calling
+        this directly.
+        """
+        self._locked = True
+
+    def _unlockFromSwitch(self):
+        """
+        Unlock this locked binary protocol so that further boxes may be sent
+        again.  This is used after an attempt to switch protocols has failed
+        for some reason.
+        """
+        if self.innerProtocol is not None:
+            raise ProtocolSwitched(
+                "Protocol already switched.  Cannot unlock.")
+        self._locked = False
+
+    def _prepareTLS(self, certificate, verifyAuthorities):
+        """
+        Used by StartTLSCommand to put us into the state where we don't
+        actually send things that get sent, instead we buffer them.  see
+        L{_sendBox}.
+        """
+        self._startingTLSBuffer = []
+        if self.hostCertificate is not None:
+            raise OnlyOneTLS(
+                "Previously authenticated connection between %s and %s "
+                "is trying to re-establish as %s" % (
+                    self.hostCertificate,
+                    self.peerCertificate,
+                    (certificate, verifyAuthorities)))
+
+    def _startTLS(self, certificate, verifyAuthorities):
+        """
+        Used by TLSBox to initiate the SSL handshake.
+
+        @param certificate: a L{twisted.internet.ssl.PrivateCertificate} for
+        use locally.
+
+        @param verifyAuthorities: L{twisted.internet.ssl.Certificate} instances
+        representing certificate authorities which will verify our peer.
+        """
+        self.hostCertificate = certificate
+        self._justStartedTLS = True
+        if verifyAuthorities is None:
+            verifyAuthorities = ()
+        self.transport.startTLS(certificate.options(*verifyAuthorities))
+        stlsb = self._startingTLSBuffer
+        if stlsb is not None:
+            self._startingTLSBuffer = None
+            for box in stlsb:
+                self.sendBox(box)
+
+    def _getPeerCertificate(self):
+        if self.noPeerCertificate:
+            return None
+        return Certificate.peerFromTransport(self.transport)
+    peerCertificate = property(_getPeerCertificate)
+
+    def unhandledError(self, failure):
+        """
+        The buck stops here.  This error was completely unhandled, time to
+        terminate the connection.
+        """
+        log.err(
+            failure,
+            "Amp server or network failure unhandled by client application.  "
+            "Dropping connection!  To avoid, add errbacks to ALL remote "
+            "commands!")
+        if self.transport is not None:
+            self.transport.loseConnection()
+
+    def _defaultStartTLSResponder(self):
+        """
+        The default TLS responder doesn't specify any certificate or anything.
+
+        From a security perspective, it's little better than a plain-text
+        connection - but it is still a *bit* better, so it's included for
+        convenience.
+
+        You probably want to override this by providing your own
+        C{StartTLS.responder}.
+        """
+        return {}
+    StartTLS.responder(_defaultStartTLSResponder)
+
+
+class AMP(BinaryBoxProtocol, BoxDispatcher,
+          CommandLocator, SimpleStringLocator):
+    """
+    This protocol is an AMP connection.  See the module docstring for protocol
+    details.
+    """
+
+    _ampInitialized = False
+
+    def __init__(self, boxReceiver=None, locator=None):
+        # For backwards compatibility.  When AMP did not separate parsing logic
+        # (L{BinaryBoxProtocol}), request-response logic (L{BoxDispatcher}) and
+        # command routing (L{CommandLocator}), it did not have a constructor.
+        # Now it does, so old subclasses might have defined their own that did
+        # not upcall.  If this flag isn't set, we'll call the constructor in
+        # makeConnection before anything actually happens.
+        self._ampInitialized = True
+        if boxReceiver is None:
+            boxReceiver = self
+        if locator is None:
+            locator = self
+        BoxDispatcher.__init__(self, locator)
+        BinaryBoxProtocol.__init__(self, boxReceiver)
+
+    def locateResponder(self, name):
+        """
+        Unify the implementations of L{CommandLocator} and
+        L{SimpleStringLocator} to perform both kinds of dispatch, preferring
+        L{CommandLocator}.
+        """
+        firstResponder = CommandLocator.locateResponder(self, name)
+        if firstResponder is not None:
+            return firstResponder
+        secondResponder = SimpleStringLocator.locateResponder(self, name)
+        return secondResponder
+
+    def __repr__(self):
+        """
+        A verbose string representation which gives us information about this
+        AMP connection.
+        """
+        if self.innerProtocol is not None:
+            innerRepr = ' inner %r' % (self.innerProtocol,)
+        else:
+            innerRepr = ''
+        return '<%s%s at 0x%x>' % (
+            self.__class__.__name__, innerRepr, id(self))
+
+    def makeConnection(self, transport):
+        """
+        Emit a helpful log message when the connection is made.
+        """
+        if not self._ampInitialized:
+            # See comment in the constructor re: backward compatibility.  I
+            # should probably emit a deprecation warning here.
+            AMP.__init__(self)
+        # Save these so we can emit a similar log message in L{connectionLost}.
+        self._transportPeer = transport.getPeer()
+        self._transportHost = transport.getHost()
+        log.msg("%s connection established (HOST:%s PEER:%s)" % (
+                self.__class__.__name__,
+                self._transportHost,
+                self._transportPeer))
+        BinaryBoxProtocol.makeConnection(self, transport)
+
+    def connectionLost(self, reason):
+        """
+        Emit a helpful log message when the connection is lost.
+        """
+        log.msg("%s connection lost (HOST:%s PEER:%s)" %
+                (self.__class__.__name__,
+                 self._transportHost,
+                 self._transportPeer))
+        BinaryBoxProtocol.connectionLost(self, reason)
+        self.transport = None
+
+
+class _ParserHelper:
+    """
+    A box receiver which records all boxes received.
+    """
+
+    def __init__(self):
+        self.boxes = []
+
+    def getPeer(self):
+        return 'string'
+
+    def getHost(self):
+        return 'string'
+
+    disconnecting = False
+
+    def startReceivingBoxes(self, sender):
+        """
+        No initialization is required.
+        """
+
+    def ampBoxReceived(self, box):
+        self.boxes.append(box)
+
+    # Synchronous helpers
+    @classmethod
+    def parse(cls, fileObj):
+        """
+        Parse some amp data stored in a file.
+
+        @param fileObj: a file-like object.
+
+        @return: a list of AmpBoxes encoded in the given file.
+        """
+        parserHelper = cls()
+        bbp = BinaryBoxProtocol(boxReceiver=parserHelper)
+        bbp.makeConnection(parserHelper)
+        bbp.dataReceived(fileObj.read())
+        return parserHelper.boxes
+
+    @classmethod
+    def parseString(cls, data):
+        """
+        Parse some amp data stored in a string.
+
+        @param data: a bytes holding some amp-encoded data.
+
+        @return: a list of AmpBoxes encoded in the given string.
+        """
+        return cls.parse(BytesIO(data))
+
+
+parse = _ParserHelper.parse
+parseString = _ParserHelper.parseString
+
+
+def _stringsToObjects(strings, arglist, proto):
+    """
+    Convert an AmpBox to a dictionary of python objects, converting through a
+    given arglist.
+
+    @param strings: an AmpBox (or dict of strings)
+
+    @param arglist: a list of 2-tuples of strings and Argument objects, as
+    described in L{Command.arguments}.
+
+    @param proto: an L{AMP} instance.
+
+    @return: the converted dictionary mapping names to argument objects.
+    """
+    objects = {}
+    myStrings = strings.copy()
+    for argname, argparser in arglist:
+        argparser.fromBox(argname, myStrings, objects, proto)
+    return objects
+
+
+def _objectsToStrings(objects, arglist, strings, proto):
+    """
+    Convert a dictionary of python objects to an AmpBox, converting through a
+    given arglist.
+
+    @param objects: a dict mapping names to python objects
+
+    @param arglist: a list of 2-tuples of strings and Argument objects, as
+    described in L{Command.arguments}.
+
+    @param strings: [OUT PARAMETER] An object providing the L{dict}
+    interface which will be populated with serialized data.
+
+    @param proto: an L{AMP} instance.
+
+    @return: The converted dictionary mapping names to encoded argument
+    strings (identical to C{strings}).
+    """
+    myObjects = objects.copy()
+    for argname, argparser in arglist:
+        argparser.toBox(argname, strings, myObjects, proto)
+    return strings
+
+
+class _FixedOffsetTZInfo(datetime.tzinfo):
+    """
+    Represents a fixed timezone offset (without daylight saving time).
+
+    @ivar name: A C{bytes} giving the name of this timezone; the name just
+        includes how much time this offset represents.
+
+    @ivar offset: A C{datetime.timedelta} giving the amount of time this
+        timezone is offset.
+    """
+
+    def __init__(self, sign, hours, minutes):
+        self.name = '%s%02i:%02i' % (sign, hours, minutes)
+        if sign == '-':
+            hours = -hours
+            minutes = -minutes
+        elif sign != '+':
+            raise ValueError('invalid sign for timezone %r' % (sign,))
+        self.offset = datetime.timedelta(hours=hours, minutes=minutes)
+
+    def utcoffset(self, dt):
+        """
+        Return this timezone's offset from UTC.
+        """
+        return self.offset
+
+    def dst(self, dt):
+        """
+        Return a zero C{datetime.timedelta} for the daylight saving time
+        offset, since there is never one.
+        """
+        return datetime.timedelta(0)
+
+    def tzname(self, dt):
+        """
+        Return a string describing this timezone.
+        """
+        return self.name
+
+
+utc = _FixedOffsetTZInfo('+', 0, 0)
+
+
+class Decimal(Argument):
+    """
+    Encodes C{decimal.Decimal} instances.
+
+    There are several ways in which a decimal value might be encoded.
+
+    Special values are encoded as special strings::
+
+      - Positive infinity is encoded as C{"Infinity"}
+      - Negative infinity is encoded as C{"-Infinity"}
+      - Quiet not-a-number is encoded as either C{"NaN"} or C{"-NaN"}
+      - Signalling not-a-number is encoded as either C{"sNaN"} or C{"-sNaN"}
+
+    Normal values are encoded using the base ten string representation, using
+    engineering notation to indicate magnitude without precision, and "normal"
+    digits to indicate precision.  For example::
+
+      - C{"1"} represents the value I{1} with precision to one place.
+      - C{"-1"} represents the value I{-1} with precision to one place.
+      - C{"1.0"} represents the value I{1} with precision to two places.
+      - C{"10"} represents the value I{10} with precision to two places.
+      - C{"1E+2"} represents the value I{10} with precision to one place.
+      - C{"1E-1"} represents the value I{0.1} with precision to one place.
+      - C{"1.5E+2"} represents the value I{15} with precision to two places.
+
+    U{http://speleotrove.com/decimal/} should be considered the authoritative
+    specification for the format.
+    """
+
+    fromString = decimal.Decimal
+
+    def toString(self, inObject):
+        """
+        Serialize a C{decimal.Decimal} instance to the specified wire format.
+        """
+        if isinstance(inObject, decimal.Decimal):
+            # Hopefully decimal.Decimal.__bytes__ actually does what we want.
+            return bytes(inObject)
+        raise ValueError(
+            "amp32.Decimal can only encode instances of decimal.Decimal")
+
+
+class DateTime(Argument):
+    """
+    Encodes C{datetime.datetime} instances.
+
+    Wire format: '%04i-%02i-%02iT%02i:%02i:%02i.%06i%s%02i:%02i'. Fields in
+    order are: year, month, day, hour, minute, second, microsecond, timezone
+    direction (+ or -), timezone hour, timezone minute. Encoded string is
+    always exactly 32 characters long. This format is compatible with ISO 8601,
+    but that does not mean all ISO 8601 dates can be accepted.
+
+    Also, note that the datetime module's notion of a "timezone" can be
+    complex, but the wire format includes only a fixed offset, so the
+    conversion is not lossless. A lossless transmission of a C{datetime}
+    instance is not feasible since the receiving end would require a Python
+    interpreter.
+
+    @ivar _positions: A sequence of slices giving the positions of various
+        interesting parts of the wire format.
+    """
+
+    _positions = [
+        slice(0, 4), slice(5, 7), slice(8, 10),  # year, month, day
+        slice(11, 13), slice(14, 16), slice(17, 19),  # hour, minute, second
+        slice(20, 26),  # microsecond
+        # intentionally skip timezone direction, as it is not an integer
+        slice(27, 29), slice(30, 32)  # timezone hour, timezone minute
+    ]
+
+    def fromString(self, s):
+        """
+        Parse a string containing a date and time in the wire format into a
+        C{datetime.datetime} instance.
+        """
+        if len(s) != 32:
+            raise ValueError('invalid date format %r' % (s,))
+
+        values = [int(s[p]) for p in self._positions]
+        sign = s[26]
+        timezone = _FixedOffsetTZInfo(sign, *values[7:])
+        values[7:] = [timezone]
+        return datetime.datetime(*values)
+
+    def toString(self, i):
+        """
+        Serialize a C{datetime.datetime} instance to a string in the specified
+        wire format.
+        """
+        offset = i.utcoffset()
+        if offset is None:
+            raise ValueError(
+                'amp32.DateTime cannot serialize naive datetime instances.  '
+                'You may find amp32.utc useful.')
+
+        minutesOffset = (offset.days * 86400 + offset.seconds) // 60
+
+        if minutesOffset > 0:
+            sign = '+'
+        else:
+            sign = '-'
+
+        # strftime has no way to format the microseconds, or put a ':' in the
+        # timezone. Suprise!
+
+        return '%04i-%02i-%02iT%02i:%02i:%02i.%06i%s%02i:%02i' % (
+            i.year,
+            i.month,
+            i.day,
+            i.hour,
+            i.minute,
+            i.second,
+            i.microsecond,
+            sign,
+            abs(minutesOffset) // 60,
+            abs(minutesOffset) % 60)
 
=== modified file 'src/provisioningserver/rpc/arguments.py'
--- src/provisioningserver/rpc/arguments.py	2014-10-30 11:22:41 +0000
+++ src/provisioningserver/rpc/arguments.py	2014-11-12 12:20:32 +0000
@@ -25,10 +25,10 @@
 import zlib
 
 from apiclient.utils import ascii_url
-from twisted.protocols import amp
-
-
-class Bytes(amp.Argument):
+from provisioningserver.rpc import amp32
+
+
+class Bytes(amp32.Argument):
     """Encode a structure on the wire as bytes.
 
     In truth, this does nothing more than assert that the inputs are
@@ -41,11 +41,11 @@
         return inObject
 
     def fromString(self, inString):
-        # inString is always a byte string, as defined by amp.Argument.
+        # inString is always a byte string, as defined by amp32.Argument.
         return inString
 
 
-class Choice(amp.Argument):
+class Choice(amp32.Argument):
     """Encode a choice to a predefined bytestring on the wire."""
 
     def __init__(self, choices, optional=False):
@@ -76,7 +76,7 @@
         return self._decode[inString]
 
 
-class ParsedURL(amp.Argument):
+class ParsedURL(amp32.Argument):
     """Encode a URL on the wire.
 
     The URL should be an instance of :py:class:`~urlparse.ParseResult`
@@ -105,11 +105,11 @@
         return urlparse.urlparse(inString)
 
 
-class StructureAsJSON(amp.Argument):
+class StructureAsJSON(amp32.Argument):
     """Encode a structure on the wire as JSON, compressed with zlib.
 
     The compressed size of the structure should not exceed
-    :py:data:`~twisted.protocols.amp.MAX_VALUE_LENGTH`, or ``0xffff``
+    :py:data:`~provisioningserver.rpc.amp32.MAX_VALUE_LENGTH`, or ``0xffff``
     bytes. This is pretty hard to be sure of ahead of time, so only use
     this for small structures that won't go near the limit.
     """
@@ -121,8 +121,8 @@
         return json.loads(zlib.decompress(inString))
 
 
-class CompressedAmpList(amp.AmpList):
-    """An :py:class:`amp.AmpList` that's compressed on the wire.
+class CompressedAmpList(amp32.AmpList):
+    """An :py:class:`amp32.AmpList` that's compressed on the wire.
 
     The serialised form is transparently compressed and decompressed with
     zlib. This can be useful when there's a lot of repetition in the list
 
=== modified file 'src/provisioningserver/rpc/cluster.py'
--- src/provisioningserver/rpc/cluster.py	2014-10-23 18:21:41 +0000
+++ src/provisioningserver/rpc/cluster.py	2014-11-12 12:20:32 +0000
@@ -37,7 +37,10 @@
     PowerActionFail,
     UnknownPowerType,
     )
-from provisioningserver.rpc import exceptions
+from provisioningserver.rpc import (
+    amp32,
+    exceptions,
+    )
 from provisioningserver.rpc.arguments import (
     Bytes,
     ParsedURL,
@@ -47,10 +50,9 @@
     Authenticate,
     Identify,
     )
-from twisted.protocols import amp
-
-
-class ListBootImages(amp.Command):
+
+
+class ListBootImages(amp32.Command):
     """List the boot images available on this cluster controller.
 
     :since: 1.5
@@ -58,20 +60,20 @@
 
     arguments = []
     response = [
-        (b"images", amp.AmpList(
-            [(b"osystem", amp.Unicode()),
-             (b"architecture", amp.Unicode()),
-             (b"subarchitecture", amp.Unicode()),
-             (b"release", amp.Unicode()),
-             (b"label", amp.Unicode()),
-             (b"purpose", amp.Unicode()),
-             (b"xinstall_type", amp.Unicode()),
-             (b"xinstall_path", amp.Unicode())]))
+        (b"images", amp32.AmpList(
+            [(b"osystem", amp32.Unicode()),
+             (b"architecture", amp32.Unicode()),
+             (b"subarchitecture", amp32.Unicode()),
+             (b"release", amp32.Unicode()),
+             (b"label", amp32.Unicode()),
+             (b"purpose", amp32.Unicode()),
+             (b"xinstall_type", amp32.Unicode()),
+             (b"xinstall_path", amp32.Unicode())]))
     ]
     errors = []
 
 
-class DescribePowerTypes(amp.Command):
+class DescribePowerTypes(amp32.Command):
     """Get a JSON Schema describing this cluster's power types.
 
     :since: 1.5
@@ -84,7 +86,7 @@
     errors = []
 
 
-class ListSupportedArchitectures(amp.Command):
+class ListSupportedArchitectures(amp32.Command):
     """Report the cluster's supported architectures.
 
     :since: 1.5
@@ -92,15 +94,15 @@
 
     arguments = []
     response = [
-        (b"architectures", amp.AmpList([
-            (b"name", amp.Unicode()),
-            (b"description", amp.Unicode()),
+        (b"architectures", amp32.AmpList([
+            (b"name", amp32.Unicode()),
+            (b"description", amp32.Unicode()),
             ])),
     ]
     errors = []
 
 
-class ListOperatingSystems(amp.Command):
+class ListOperatingSystems(amp32.Command):
     """Report the cluster's supported operating systems.
 
     :since: 1.7
@@ -108,34 +110,34 @@
 
     arguments = []
     response = [
-        (b"osystems", amp.AmpList([
-            (b"name", amp.Unicode()),
-            (b"title", amp.Unicode()),
-            (b"releases", amp.AmpList([
-                (b"name", amp.Unicode()),
-                (b"title", amp.Unicode()),
-                (b"requires_license_key", amp.Boolean()),
-                (b"can_commission", amp.Boolean()),
+        (b"osystems", amp32.AmpList([
+            (b"name", amp32.Unicode()),
+            (b"title", amp32.Unicode()),
+            (b"releases", amp32.AmpList([
+                (b"name", amp32.Unicode()),
+                (b"title", amp32.Unicode()),
+                (b"requires_license_key", amp32.Boolean()),
+                (b"can_commission", amp32.Boolean()),
             ])),
-            (b"default_release", amp.Unicode(optional=True)),
-            (b"default_commissioning_release", amp.Unicode(optional=True)),
+            (b"default_release", amp32.Unicode(optional=True)),
+            (b"default_commissioning_release", amp32.Unicode(optional=True)),
         ])),
     ]
     errors = []
 
 
-class GetOSReleaseTitle(amp.Command):
+class GetOSReleaseTitle(amp32.Command):
     """Get the title for the operating systems release.
 
     :since: 1.7
     """
 
     arguments = [
-        (b"osystem", amp.Unicode()),
-        (b"release", amp.Unicode()),
+        (b"osystem", amp32.Unicode()),
+        (b"release", amp32.Unicode()),
     ]
     response = [
-        (b"title", amp.Unicode()),
+        (b"title", amp32.Unicode()),
     ]
     errors = {
         exceptions.NoSuchOperatingSystem: (
@@ -143,19 +145,19 @@
     }
 
 
-class ValidateLicenseKey(amp.Command):
+class ValidateLicenseKey(amp32.Command):
     """Validate an OS license key.
 
     :since: 1.7
     """
 
     arguments = [
-        (b"osystem", amp.Unicode()),
-        (b"release", amp.Unicode()),
-        (b"key", amp.Unicode()),
+        (b"osystem", amp32.Unicode()),
+        (b"release", amp32.Unicode()),
+        (b"key", amp32.Unicode()),
     ]
     response = [
-        (b"is_valid", amp.Boolean()),
+        (b"is_valid", amp32.Boolean()),
     ]
     errors = {
         exceptions.NoSuchOperatingSystem: (
@@ -163,20 +165,20 @@
     }
 
 
-class GetPreseedData(amp.Command):
+class GetPreseedData(amp32.Command):
     """Get OS-specific preseed data.
 
     :since: 1.7
     """
 
     arguments = [
-        (b"osystem", amp.Unicode()),
-        (b"preseed_type", amp.Unicode()),
-        (b"node_system_id", amp.Unicode()),
-        (b"node_hostname", amp.Unicode()),
-        (b"consumer_key", amp.Unicode()),
-        (b"token_key", amp.Unicode()),
-        (b"token_secret", amp.Unicode()),
+        (b"osystem", amp32.Unicode()),
+        (b"preseed_type", amp32.Unicode()),
+        (b"node_system_id", amp32.Unicode()),
+        (b"node_hostname", amp32.Unicode()),
+        (b"consumer_key", amp32.Unicode()),
+        (b"token_key", amp32.Unicode()),
+        (b"token_secret", amp32.Unicode()),
         (b"metadata_url", ParsedURL()),
     ]
     response = [
@@ -190,16 +192,16 @@
     }
 
 
-class ComposeCurtinNetworkPreseed(amp.Command):
+class ComposeCurtinNetworkPreseed(amp32.Command):
     """Generate Curtin network preseed for a node.
 
     :since: 1.7
     """
 
     arguments = [
-        (b"osystem", amp.Unicode()),
+        (b"osystem", amp32.Unicode()),
         (b"config", StructureAsJSON()),
-        (b"disable_ipv4", amp.Boolean()),
+        (b"disable_ipv4", amp32.Boolean()),
         ]
     response = [
         (b"data", StructureAsJSON()),
@@ -209,16 +211,16 @@
     }
 
 
-class _Power(amp.Command):
+class _Power(amp32.Command):
     """Base class for power control commands.
 
     :since: 1.7
     """
 
     arguments = [
-        (b"system_id", amp.Unicode()),
-        (b"hostname", amp.Unicode()),
-        (b"power_type", amp.Unicode()),
+        (b"system_id", amp32.Unicode()),
+        (b"hostname", amp32.Unicode()),
+        (b"power_type", amp32.Unicode()),
         # We can't define a tighter schema here because this is a highly
         # variable bag of arguments from a variety of sources.
         (b"context", StructureAsJSON()),
@@ -256,29 +258,29 @@
     :since: 1.7
     """
     response = [
-        (b"state", amp.Unicode()),
+        (b"state", amp32.Unicode()),
     ]
 
 
-class _ConfigureDHCP(amp.Command):
+class _ConfigureDHCP(amp32.Command):
     """Configure a DHCP server.
 
     :since: 1.7
     """
     arguments = [
-        (b"omapi_key", amp.Unicode()),
-        (b"subnet_configs", amp.AmpList([
-            (b"subnet", amp.Unicode()),
-            (b"subnet_mask", amp.Unicode()),
-            (b"subnet_cidr", amp.Unicode()),
-            (b"broadcast_ip", amp.Unicode()),
-            (b"interface", amp.Unicode()),
-            (b"router_ip", amp.Unicode()),
-            (b"dns_servers", amp.Unicode()),
-            (b"ntp_server", amp.Unicode()),
-            (b"domain_name", amp.Unicode()),
-            (b"ip_range_low", amp.Unicode()),
-            (b"ip_range_high", amp.Unicode()),
+        (b"omapi_key", amp32.Unicode()),
+        (b"subnet_configs", amp32.AmpList([
+            (b"subnet", amp32.Unicode()),
+            (b"subnet_mask", amp32.Unicode()),
+            (b"subnet_cidr", amp32.Unicode()),
+            (b"broadcast_ip", amp32.Unicode()),
+            (b"interface", amp32.Unicode()),
+            (b"router_ip", amp32.Unicode()),
+            (b"dns_servers", amp32.Unicode()),
+            (b"ntp_server", amp32.Unicode()),
+            (b"domain_name", amp32.Unicode()),
+            (b"ip_range_low", amp32.Unicode()),
+            (b"ip_range_high", amp32.Unicode()),
             ])),
         ]
     response = []
@@ -299,18 +301,18 @@
     """
 
 
-class CreateHostMaps(amp.Command):
+class CreateHostMaps(amp32.Command):
     """Create host maps in the DHCP server's configuration.
 
     :since: 1.7
     """
 
     arguments = [
-        (b"mappings", amp.AmpList([
-            (b"ip_address", amp.Unicode()),
-            (b"mac_address", amp.Unicode()),
+        (b"mappings", amp32.AmpList([
+            (b"ip_address", amp32.Unicode()),
+            (b"mac_address", amp32.Unicode()),
         ])),
-        (b"shared_key", amp.Unicode()),
+        (b"shared_key", amp32.Unicode()),
     ]
     response = []
     errors = {
@@ -319,15 +321,15 @@
     }
 
 
-class RemoveHostMaps(amp.Command):
+class RemoveHostMaps(amp32.Command):
     """Remove host maps from the DHCP server's configuration.
 
     :since: 1.7
     """
 
     arguments = [
-        (b"ip_addresses", amp.ListOf(amp.Unicode())),
-        (b"shared_key", amp.Unicode()),
+        (b"ip_addresses", amp32.ListOf(amp32.Unicode())),
+        (b"shared_key", amp32.Unicode()),
     ]
     response = []
     errors = {
@@ -336,7 +338,7 @@
     }
 
 
-class ImportBootImages(amp.Command):
+class ImportBootImages(amp32.Command):
     """Import boot images and report the final
     boot images that exist on the cluster.
 
@@ -344,15 +346,15 @@
     """
 
     arguments = [
-        (b"sources", amp.AmpList(
-            [(b"url", amp.Unicode()),
+        (b"sources", amp32.AmpList(
+            [(b"url", amp32.Unicode()),
              (b"keyring_data", Bytes()),
-             (b"selections", amp.AmpList(
-                 [(b"os", amp.Unicode()),
-                  (b"release", amp.Unicode()),
-                  (b"arches", amp.ListOf(amp.Unicode())),
-                  (b"subarches", amp.ListOf(amp.Unicode())),
-                  (b"labels", amp.ListOf(amp.Unicode()))]))])),
+             (b"selections", amp32.AmpList(
+                 [(b"os", amp32.Unicode()),
+                  (b"release", amp32.Unicode()),
+                  (b"arches", amp32.ListOf(amp32.Unicode())),
+                  (b"subarches", amp32.ListOf(amp32.Unicode())),
+                  (b"labels", amp32.ListOf(amp32.Unicode()))]))])),
         (b"http_proxy", ParsedURL(optional=True)),
         (b"https_proxy", ParsedURL(optional=True)),
     ]
@@ -360,80 +362,80 @@
     errors = []
 
 
-class StartMonitors(amp.Command):
+class StartMonitors(amp32.Command):
     """Starts monitors(s) on the cluster.
 
     :since: 1.7
     """
 
     arguments = [
-        (b"monitors", amp.AmpList(
-            [(b"deadline", amp.DateTime()),
+        (b"monitors", amp32.AmpList(
+            [(b"deadline", amp32.DateTime()),
              (b"context", StructureAsJSON()),
-             (b"id", amp.Unicode()),
+             (b"id", amp32.Unicode()),
              ]))
     ]
     response = []
     errors = []
 
 
-class CancelMonitor(amp.Command):
+class CancelMonitor(amp32.Command):
     """Cancels an existing monitor on the cluster.
 
     :since: 1.7
     """
 
     arguments = [
-        (b"id", amp.Unicode()),
+        (b"id", amp32.Unicode()),
         ]
     response = []
     error = []
 
 
-class EvaluateTag(amp.Command):
+class EvaluateTag(amp32.Command):
     """Evaluate a tag against all of the cluster's nodes.
 
     :since: 1.7
     """
 
     arguments = [
-        (b"tag_name", amp.Unicode()),
-        (b"tag_definition", amp.Unicode()),
-        (b"tag_nsmap", amp.AmpList([
-            (b"prefix", amp.Unicode()),
-            (b"uri", amp.Unicode()),
+        (b"tag_name", amp32.Unicode()),
+        (b"tag_definition", amp32.Unicode()),
+        (b"tag_nsmap", amp32.AmpList([
+            (b"prefix", amp32.Unicode()),
+            (b"uri", amp32.Unicode()),
         ])),
         # A 3-part credential string for the web API.
-        (b"credentials", amp.Unicode()),
+        (b"credentials", amp32.Unicode()),
     ]
     response = []
     errors = []
 
 
-class AddVirsh(amp.Command):
+class AddVirsh(amp32.Command):
     """Probe for and enlist virsh VMs attached to the cluster.
 
     :since: 1.7
     """
 
     arguments = [
-        (b"poweraddr", amp.Unicode()),
-        (b"password", amp.Unicode(optional=True)),
+        (b"poweraddr", amp32.Unicode()),
+        (b"password", amp32.Unicode(optional=True)),
     ]
     response = []
     errors = []
 
 
-class AddSeaMicro15k(amp.Command):
+class AddSeaMicro15k(amp32.Command):
     """Probe for and enlist seamicro15k machines attached to the cluster.
 
     :since: 1.7
     """
     arguments = [
-        (b"mac", amp.Unicode()),
-        (b"username", amp.Unicode()),
-        (b"password", amp.Unicode()),
-        (b"power_control", amp.Unicode(optional=True)),
+        (b"mac", amp32.Unicode()),
+        (b"username", amp32.Unicode()),
+        (b"password", amp32.Unicode()),
+        (b"power_control", amp32.Unicode(optional=True)),
     ]
     response = []
     errors = {
@@ -441,41 +443,41 @@
     }
 
 
-class EnlistNodesFromMSCM(amp.Command):
+class EnlistNodesFromMSCM(amp32.Command):
     """Probe for and enlist mscm machines attached to the cluster.
 
     :since: 1.7
     """
     arguments = [
-        (b"host", amp.Unicode()),
-        (b"username", amp.Unicode()),
-        (b"password", amp.Unicode()),
+        (b"host", amp32.Unicode()),
+        (b"username", amp32.Unicode()),
+        (b"password", amp32.Unicode()),
     ]
     response = []
     errors = {}
 
 
-class EnlistNodesFromUCSM(amp.Command):
+class EnlistNodesFromUCSM(amp32.Command):
     """Probe for and enlist ucsm machines attached to the cluster.
 
     :since: 1.7
     """
     arguments = [
-        (b"url", amp.Unicode()),
-        (b"username", amp.Unicode()),
-        (b"password", amp.Unicode()),
+        (b"url", amp32.Unicode()),
+        (b"username", amp32.Unicode()),
+        (b"password", amp32.Unicode()),
     ]
     response = []
     errors = {}
 
 
-class IsImportBootImagesRunning(amp.Command):
+class IsImportBootImagesRunning(amp32.Command):
     """Check if the import boot images task is running on the cluster.
 
     :since: 1.7
     """
     arguments = []
     response = [
-        (b"running", amp.Boolean()),
+        (b"running", amp32.Boolean()),
     ]
     errors = {}
 
=== modified file 'src/provisioningserver/rpc/clusterservice.py'
--- src/provisioningserver/rpc/clusterservice.py	2014-11-10 15:11:58 +0000
+++ src/provisioningserver/rpc/clusterservice.py	2014-11-12 12:20:32 +0000
@@ -44,6 +44,7 @@
 from provisioningserver.logger.utils import log_call
 from provisioningserver.network import discover_networks
 from provisioningserver.rpc import (
+    amp32,
     cluster,
     common,
     dhcp,
@@ -97,7 +98,6 @@
     ConnectionClosed,
     )
 from twisted.internet.threads import deferToThread
-from twisted.protocols import amp
 from twisted.python import log
 from twisted.web import http
 import twisted.web.client
@@ -312,11 +312,11 @@
         cancel_monitor(id)
         return {}
 
-    @amp.StartTLS.responder
+    @amp32.StartTLS.responder
     def get_tls_parameters(self):
         """get_tls_parameters()
 
-        Implementation of :py:class:`~twisted.protocols.amp.StartTLS`.
+        Implementation of :py:class:`~provisioningserver.rpc.amp32.StartTLS`.
         """
         try:
             from provisioningserver.rpc.testing import tls
@@ -551,7 +551,7 @@
 
     @inlineCallbacks
     def secureConnection(self):
-        yield self.callRemote(amp.StartTLS, **self.get_tls_parameters())
+        yield self.callRemote(amp32.StartTLS, **self.get_tls_parameters())
 
         # For some weird reason (it's mentioned in Twisted's source),
         # TLS negotiation does not complete until we do something with
 
=== modified file 'src/provisioningserver/rpc/common.py'
--- src/provisioningserver/rpc/common.py	2014-10-03 12:52:35 +0000
+++ src/provisioningserver/rpc/common.py	2014-11-12 12:20:32 +0000
@@ -19,22 +19,22 @@
     "RPCProtocol",
 ]
 
+from provisioningserver.rpc import amp32
 from provisioningserver.rpc.interfaces import IConnection
 from provisioningserver.utils.twisted import asynchronous
 from twisted.internet.defer import Deferred
-from twisted.protocols import amp
-
-
-class Identify(amp.Command):
+
+
+class Identify(amp32.Command):
     """Request the identity of the remote side, e.g. its UUID.
 
     :since: 1.5
     """
 
-    response = [(b"ident", amp.Unicode())]
-
-
-class Authenticate(amp.Command):
+    response = [(b"ident", amp32.Unicode())]
+
+
+class Authenticate(amp32.Command):
     """Authenticate the remote side.
 
     The procedure is as follows:
@@ -63,19 +63,19 @@
     """
 
     arguments = [
-        (b"message", amp.String()),
+        (b"message", amp32.String()),
     ]
     response = [
-        (b"digest", amp.String()),
-        (b"salt", amp.String()),  # Is 'salt' the right term here?
+        (b"digest", amp32.String()),
+        (b"salt", amp32.String()),  # Is 'salt' the right term here?
     ]
     errors = []
 
 
 class Client:
-    """Wrapper around an :class:`amp.AMP` instance.
+    """Wrapper around an :class:`amp32.AMP` instance.
 
-    Limits the API to a subset of the behaviour of :class:`amp.AMP`'s,
+    Limits the API to a subset of the behaviour of :class:`amp32.AMP`'s,
     with alterations to make it suitable for use from a thread outside
     of the reactor.
     """
@@ -107,7 +107,7 @@
             different stack from the caller's, e.g. when calling into the
             Twisted reactor from a thread.
 
-        :param cmd: The `amp.Command` child class representing the remote
+        :param cmd: The `amp32.Command` child class representing the remote
             method to be invoked.
         :param kwargs: Any parameters to the remote method.  Only keyword
             arguments are accepted.
@@ -143,10 +143,10 @@
         return hash(self._conn)
 
 
-class RPCProtocol(amp.AMP, object):
-    """A specialisation of `amp.AMP`.
+class RPCProtocol(amp32.AMP, object):
+    """A specialisation of `amp32.AMP`.
 
-    It's hard to track exactly when an `amp.AMP` protocol is connected to its
+    It's hard to track exactly when an `amp32.AMP` protocol is connected to its
     transport, or disconnected, from the "outside". It's necessary to subclass
     and override `connectionMade` and `connectionLost` and signal from there,
     which is what this class does.
 
=== modified file 'src/provisioningserver/rpc/monitors.py'
--- src/provisioningserver/rpc/monitors.py	2014-10-16 11:12:48 +0000
+++ src/provisioningserver/rpc/monitors.py	2014-11-12 12:20:32 +0000
@@ -20,11 +20,13 @@
 from datetime import datetime
 
 from provisioningserver.logger import get_maas_logger
-from provisioningserver.rpc import getRegionClient
+from provisioningserver.rpc import (
+    amp32,
+    getRegionClient,
+    )
 from provisioningserver.rpc.exceptions import NoConnectionsAvailable
 from provisioningserver.rpc.region import MonitorExpired
 from twisted.internet import reactor
-from twisted.protocols import amp
 
 
 maaslog = get_maas_logger("monitors")
@@ -47,7 +49,7 @@
     monitor ID.
     """
     for monitor in monitors:
-        delay = monitor["deadline"] - datetime.now(amp.utc)
+        delay = monitor["deadline"] - datetime.now(amp32.utc)
         monitor_id = monitor["id"]
         if monitor_id in running_monitors:
             dc, _ = running_monitors.pop(monitor_id)
 
=== modified file 'src/provisioningserver/rpc/region.py'
--- src/provisioningserver/rpc/region.py	2014-10-30 11:29:22 +0000
+++ src/provisioningserver/rpc/region.py	2014-11-12 12:20:32 +0000
@@ -40,6 +40,7 @@
     "UpdateNodePowerState",
 ]
 
+from provisioningserver.rpc import amp32
 from provisioningserver.rpc.arguments import (
     Bytes,
     Choice,
@@ -59,10 +60,9 @@
     NoSuchEventType,
     NoSuchNode,
     )
-from twisted.protocols import amp
-
-
-class Register(amp.Command):
+
+
+class Register(amp32.Command):
     """Register a cluster with the region controller.
 
     This is the last part of the Authenticate and Register two-step. See
@@ -72,11 +72,11 @@
     """
 
     arguments = [
-        (b"uuid", amp.Unicode()),
-        (b"networks", amp.AmpList([
-            (b"interface", amp.Unicode()),
-            (b"ip", amp.Unicode()),
-            (b"subnet_mask", amp.Unicode()),
+        (b"uuid", amp32.Unicode()),
+        (b"networks", amp32.AmpList([
+            (b"interface", amp32.Unicode()),
+            (b"ip", amp32.Unicode()),
+            (b"subnet_mask", amp32.Unicode()),
         ], optional=True)),
         # The URL for the region as seen by the cluster.
         (b"url", ParsedURL(optional=True)),
@@ -87,7 +87,7 @@
     }
 
 
-class ReportBootImages(amp.Command):
+class ReportBootImages(amp32.Command):
     """Report boot images available on the invoking cluster controller.
 
     :since: 1.5
@@ -95,18 +95,18 @@
 
     arguments = [
         # The cluster UUID.
-        (b"uuid", amp.Unicode()),
-        (b"images", amp.AmpList(
-            [(b"architecture", amp.Unicode()),
-             (b"subarchitecture", amp.Unicode()),
-             (b"release", amp.Unicode()),
-             (b"purpose", amp.Unicode())])),
+        (b"uuid", amp32.Unicode()),
+        (b"images", amp32.AmpList(
+            [(b"architecture", amp32.Unicode()),
+             (b"subarchitecture", amp32.Unicode()),
+             (b"release", amp32.Unicode()),
+             (b"purpose", amp32.Unicode())])),
     ]
     response = []
     errors = []
 
 
-class GetBootSources(amp.Command):
+class GetBootSources(amp32.Command):
     """Report boot sources and selections for the given cluster.
 
     :since: 1.6
@@ -115,22 +115,22 @@
 
     arguments = [
         # The cluster UUID.
-        (b"uuid", amp.Unicode()),
+        (b"uuid", amp32.Unicode()),
     ]
     response = [
-        (b"sources", amp.AmpList(
-            [(b"url", amp.Unicode()),
+        (b"sources", amp32.AmpList(
+            [(b"url", amp32.Unicode()),
              (b"keyring_data", Bytes()),
-             (b"selections", amp.AmpList(
-                 [(b"release", amp.Unicode()),
-                  (b"arches", amp.ListOf(amp.Unicode())),
-                  (b"subarches", amp.ListOf(amp.Unicode())),
-                  (b"labels", amp.ListOf(amp.Unicode()))]))])),
+             (b"selections", amp32.AmpList(
+                 [(b"release", amp32.Unicode()),
+                  (b"arches", amp32.ListOf(amp32.Unicode())),
+                  (b"subarches", amp32.ListOf(amp32.Unicode())),
+                  (b"labels", amp32.ListOf(amp32.Unicode()))]))])),
     ]
     errors = []
 
 
-class GetBootSourcesV2(amp.Command):
+class GetBootSourcesV2(amp32.Command):
     """Report boot sources and selections for the given cluster.
 
     Includes the new os field for the selections.
@@ -140,33 +140,33 @@
 
     arguments = [
         # The cluster UUID.
-        (b"uuid", amp.Unicode()),
+        (b"uuid", amp32.Unicode()),
     ]
     response = [
-        (b"sources", amp.AmpList(
-            [(b"url", amp.Unicode()),
+        (b"sources", amp32.AmpList(
+            [(b"url", amp32.Unicode()),
              (b"keyring_data", Bytes()),
-             (b"selections", amp.AmpList(
-                 [(b"os", amp.Unicode()),
-                  (b"release", amp.Unicode()),
-                  (b"arches", amp.ListOf(amp.Unicode())),
-                  (b"subarches", amp.ListOf(amp.Unicode())),
-                  (b"labels", amp.ListOf(amp.Unicode()))]))])),
+             (b"selections", amp32.AmpList(
+                 [(b"os", amp32.Unicode()),
+                  (b"release", amp32.Unicode()),
+                  (b"arches", amp32.ListOf(amp32.Unicode())),
+                  (b"subarches", amp32.ListOf(amp32.Unicode())),
+                  (b"labels", amp32.ListOf(amp32.Unicode()))]))])),
     ]
     errors = []
 
 
-class UpdateLeases(amp.Command):
+class UpdateLeases(amp32.Command):
     """Report DHCP leases on the invoking cluster controller.
 
     :since: 1.7
     """
     arguments = [
         # The cluster UUID.
-        (b"uuid", amp.Unicode()),
+        (b"uuid", amp32.Unicode()),
         (b"mappings", CompressedAmpList(
-            [(b"ip", amp.Unicode()),
-             (b"mac", amp.Unicode())]))
+            [(b"ip", amp32.Unicode()),
+             (b"mac", amp32.Unicode())]))
     ]
     response = []
     errors = {
@@ -174,7 +174,7 @@
     }
 
 
-class GetArchiveMirrors(amp.Command):
+class GetArchiveMirrors(amp32.Command):
     """Return the Main and Port mirrors to use.
 
     :since: 1.7
@@ -187,7 +187,7 @@
     errors = []
 
 
-class GetProxies(amp.Command):
+class GetProxies(amp32.Command):
     """Return the HTTP and HTTPS proxies to use.
 
     :since: 1.6
@@ -201,7 +201,7 @@
     errors = []
 
 
-class GetClusterStatus(amp.Command):
+class GetClusterStatus(amp32.Command):
     """Return the status of the given cluster.
 
     :since: 1.7
@@ -209,7 +209,7 @@
 
     arguments = [
         # The cluster UUID.
-        (b"uuid", amp.Unicode()),
+        (b"uuid", amp32.Unicode()),
     ]
     _response_status_choices = {
         0: b"PENDING",   # NODEGROUP_STATUS.PENDING
@@ -224,7 +224,7 @@
     }
 
 
-class MarkNodeFailed(amp.Command):
+class MarkNodeFailed(amp32.Command):
     """Mark a node as 'broken'.
 
     :since: 1.7
@@ -232,9 +232,9 @@
 
     arguments = [
         # The node's system_id.
-        (b"system_id", amp.Unicode()),
+        (b"system_id", amp32.Unicode()),
         # The error description.
-        (b"error_description", amp.Unicode()),
+        (b"error_description", amp32.Unicode()),
     ]
     response = []
     errors = {
@@ -243,7 +243,7 @@
     }
 
 
-class ListNodePowerParameters(amp.Command):
+class ListNodePowerParameters(amp32.Command):
     """Return the list of power parameters for nodes
     that this cluster controls.
 
@@ -255,14 +255,14 @@
 
     arguments = [
         # The cluster UUID.
-        (b"uuid", amp.Unicode()),
+        (b"uuid", amp32.Unicode()),
     ]
     response = [
-        (b"nodes", amp.AmpList(
-            [(b"system_id", amp.Unicode()),
-             (b"hostname", amp.Unicode()),
-             (b"power_state", amp.Unicode()),
-             (b"power_type", amp.Unicode()),
+        (b"nodes", amp32.AmpList(
+            [(b"system_id", amp32.Unicode()),
+             (b"hostname", amp32.Unicode()),
+             (b"power_state", amp32.Unicode()),
+             (b"power_type", amp32.Unicode()),
              # We can't define a tighter schema here because this is a highly
              # variable bag of arguments from a variety of sources.
              (b"context", StructureAsJSON())])),
@@ -272,7 +272,7 @@
     }
 
 
-class UpdateNodePowerState(amp.Command):
+class UpdateNodePowerState(amp32.Command):
     """Update Node Power State.
 
     :since: 1.7
@@ -280,120 +280,120 @@
 
     arguments = [
         # The node's system_id.
-        (b"system_id", amp.Unicode()),
+        (b"system_id", amp32.Unicode()),
         # The node's power_state.
-        (b"power_state", amp.Unicode()),
+        (b"power_state", amp32.Unicode()),
     ]
     response = []
     errors = {NoSuchNode: b"NoSuchNode"}
 
 
-class RegisterEventType(amp.Command):
+class RegisterEventType(amp32.Command):
     """Register an event type.
 
     :since: 1.7
     """
 
     arguments = [
-        (b"name", amp.Unicode()),
-        (b"description", amp.Unicode()),
-        (b"level", amp.Integer()),
+        (b"name", amp32.Unicode()),
+        (b"description", amp32.Unicode()),
+        (b"level", amp32.Integer()),
     ]
     response = []
     errors = []
 
 
-class SendEvent(amp.Command):
-    """Send an event.
-
-    :since: 1.7
-    """
-
-    arguments = [
-        (b"system_id", amp.Unicode()),
-        (b"type_name", amp.Unicode()),
-        (b"description", amp.Unicode()),
-    ]
-    response = []
-    errors = {
-        NoSuchNode: b"NoSuchNode",
-        NoSuchEventType: b"NoSuchEventType"
-    }
-
-
-class SendEventMACAddress(amp.Command):
-    """Send an event.
-
-    :since: 1.7
-    """
-
-    arguments = [
-        (b"mac_address", amp.Unicode()),
-        (b"type_name", amp.Unicode()),
-        (b"description", amp.Unicode()),
-    ]
-    response = []
-    errors = {
-        NoSuchNode: b"NoSuchNode",
-        NoSuchEventType: b"NoSuchEventType"
-    }
-
-
-class ReportForeignDHCPServer(amp.Command):
+class SendEvent(amp32.Command):
+    """Send an event.
+
+    :since: 1.7
+    """
+
+    arguments = [
+        (b"system_id", amp32.Unicode()),
+        (b"type_name", amp32.Unicode()),
+        (b"description", amp32.Unicode()),
+    ]
+    response = []
+    errors = {
+        NoSuchNode: b"NoSuchNode",
+        NoSuchEventType: b"NoSuchEventType"
+    }
+
+
+class SendEventMACAddress(amp32.Command):
+    """Send an event.
+
+    :since: 1.7
+    """
+
+    arguments = [
+        (b"mac_address", amp32.Unicode()),
+        (b"type_name", amp32.Unicode()),
+        (b"description", amp32.Unicode()),
+    ]
+    response = []
+    errors = {
+        NoSuchNode: b"NoSuchNode",
+        NoSuchEventType: b"NoSuchEventType"
+    }
+
+
+class ReportForeignDHCPServer(amp32.Command):
     """Report a foreign DHCP server on the cluster's network.
 
     :since: 1.7
     """
 
     arguments = [
-        (b"cluster_uuid", amp.Unicode()),
-        (b"interface_name", amp.Unicode()),
-        (b"foreign_dhcp_ip", amp.Unicode(optional=True)),
+        (b"cluster_uuid", amp32.Unicode()),
+        (b"interface_name", amp32.Unicode()),
+        (b"foreign_dhcp_ip", amp32.Unicode(optional=True)),
     ]
     response = []
     errors = []
 
 
-class GetClusterInterfaces(amp.Command):
+class GetClusterInterfaces(amp32.Command):
     """Fetch the known interfaces for a cluster from the region.
 
     :since: 1.7
     """
 
     arguments = [
-        (b"cluster_uuid", amp.Unicode()),
+        (b"cluster_uuid", amp32.Unicode()),
     ]
     response = [
-        (b"interfaces", amp.AmpList(
-            [(b"name", amp.Unicode()),
-             (b"interface", amp.Unicode()),
-             (b"ip", amp.Unicode())]))
+        (b"interfaces", amp32.AmpList(
+            [(b"name", amp32.Unicode()),
+             (b"interface", amp32.Unicode()),
+             (b"ip", amp32.Unicode())]))
     ]
     errors = []
 
 
-class CreateNode(amp.Command):
+class CreateNode(amp32.Command):
     """Create a node on a given cluster.
 
     :since: 1.7
     """
 
     arguments = [
-        (b'cluster_uuid', amp.Unicode()),
-        (b'architecture', amp.Unicode()),
-        (b'power_type', amp.Unicode()),
-        (b'power_parameters', amp.Unicode()),
-        (b'mac_addresses', amp.ListOf(amp.Unicode())),
+        (b'cluster_uuid', amp32.Unicode()),
+        (b'architecture', amp32.Unicode()),
+        (b'power_type', amp32.Unicode()),
+        (b'power_parameters', amp32.Unicode()),
+        (b'mac_addresses', amp32.ListOf(amp32.Unicode())),
     ]
     response = [
-        (b'system_id', amp.Unicode()),
+        (b'system_id', amp32.Unicode()),
     ]
     errors = {
         NodeAlreadyExists: b"NodeAlreadyExists",
     }
 
 
-class MonitorExpired(amp.Command):
+class MonitorExpired(amp32.Command):
     """Called by a cluster when a running monitor hits its deadline.
 
     The original context parameter from the StartMonitors call is returned.
@@ -402,14 +402,14 @@
     """
 
     arguments = [
-        (b"id", amp.Unicode()),
+        (b"id", amp32.Unicode()),
         (b"context", StructureAsJSON()),
         ]
     response = []
     errors = []
 
 
-class ReloadCluster(amp.Command):
+class ReloadCluster(amp32.Command):
     """Called by a cluster when it wants to reload its state.
 
     The region may respond with many different calls to the cluster
@@ -424,30 +424,30 @@
     """
 
     arguments = [
-        (b"cluster_uuid", amp.Unicode()),
+        (b"cluster_uuid", amp32.Unicode()),
         ]
     response = []
     errors = []
 
 
-class RequestNodeInfoByMACAddress(amp.Command):
+class RequestNodeInfoByMACAddress(amp32.Command):
     """Request Node information by mac address.
 
     :since: 1.7
     """
 
     arguments = [
-        (b"mac_address", amp.Unicode()),
+        (b"mac_address", amp32.Unicode()),
     ]
     response = [
-        (b"system_id", amp.Unicode()),
-        (b"hostname", amp.Unicode()),
-        (b"status", amp.Integer()),
-        (b"boot_type", amp.Unicode()),
-        (b"osystem", amp.Unicode()),
-        (b"distro_series", amp.Unicode()),
-        (b"architecture", amp.Unicode()),
-        (b"purpose", amp.Unicode()),
+        (b"system_id", amp32.Unicode()),
+        (b"hostname", amp32.Unicode()),
+        (b"status", amp32.Integer()),
+        (b"boot_type", amp32.Unicode()),
+        (b"osystem", amp32.Unicode()),
+        (b"distro_series", amp32.Unicode()),
+        (b"architecture", amp32.Unicode()),
+        (b"purpose", amp32.Unicode()),
     ]
     errors = {
         NoSuchNode: b"NoSuchNode",
 
=== modified file 'src/provisioningserver/rpc/testing/__init__.py'
--- src/provisioningserver/rpc/testing/__init__.py	2014-10-10 16:44:15 +0000
+++ src/provisioningserver/rpc/testing/__init__.py	2014-11-12 12:20:32 +0000
@@ -45,7 +45,10 @@
     sentinel,
     )
 import provisioningserver
-from provisioningserver.rpc import region
+from provisioningserver.rpc import (
+    amp32,
+    region,
+    )
 from provisioningserver.rpc.clusterservice import (
     Cluster,
     ClusterClient,
@@ -80,7 +83,6 @@
     )
 from twisted.internet.protocol import Factory
 from twisted.internet.task import Clock
-from twisted.protocols import amp
 from twisted.python import (
     log,
     reflect,
@@ -100,11 +102,11 @@
     d.addCallback(command.parseResponse, protocol)
 
     def eb_massage_error(error):
-        if error.check(amp.RemoteAmpError):
+        if error.check(amp32.RemoteAmpError):
             # Convert remote errors back into local errors using the
             # command's error map if possible.
             error_type = command.reverseErrors.get(
-                error.value.errorCode, amp.UnknownRemoteError)
+                error.value.errorCode, amp32.UnknownRemoteError)
             return Failure(error_type(error.value.description))
         else:
             # Exceptions raised in responders that aren't declared in that
@@ -112,7 +114,7 @@
             # in RemoteAmpError. This is because call_responder() bypasses the
             # network marshall/unmarshall steps, where these exceptions would
             # ordinarily get squashed.
-            return Failure(amp.UnknownRemoteError("%s: %s" % (
+            return Failure(amp32.UnknownRemoteError("%s: %s" % (
                 reflect.qual(error.type), reflect.safe_str(error.value))))
     d.addErrback(eb_massage_error)
 
@@ -261,7 +263,7 @@
     def addEventLoop(self, protocol):
         """Add a new stub event-loop using the given `protocol`.
 
-        The `protocol` should be an instance of `amp.AMP`.
+        The `protocol` should be an instance of `amp32.AMP`.
 
         :returns: py:class:`twisted.test.iosim.IOPump`
         """
@@ -281,8 +283,8 @@
             commands = commands + (region.Authenticate,)
         if region.Register not in commands:
             commands = commands + (region.Register,)
-        if amp.StartTLS not in commands:
-            commands = commands + (amp.StartTLS,)
+        if amp32.StartTLS not in commands:
+            commands = commands + (amp32.StartTLS,)
         protocol_factory = make_amp_protocol_factory(*commands)
         protocol = protocol_factory()
         eventloop = self.getEventLoopName(protocol)
 
=== modified file 'src/provisioningserver/rpc/testing/tls.py'
--- src/provisioningserver/rpc/testing/tls.py	2014-09-03 12:23:11 +0000
+++ src/provisioningserver/rpc/testing/tls.py	2014-11-12 12:20:32 +0000
@@ -26,7 +26,7 @@
 def get_tls_parameters(private_cert_name, trust_cert_name):
     """get_tls_parameters()
 
-    Implementation of :py:class:`~twisted.protocols.amp.StartTLS`.
+    Implementation of :py:class:`~provisioningserver.rpc.amp32.StartTLS`.
     """
     testing = filepath.FilePath(__file__).parent()
     with testing.child(private_cert_name).open() as fin:
 
=== added file 'src/provisioningserver/rpc/tests/test_amp32.py'
--- src/provisioningserver/rpc/tests/test_amp32.py	1970-01-01 00:00:00 +0000
+++ src/provisioningserver/rpc/tests/test_amp32.py	2014-11-12 12:20:32 +0000
@@ -0,0 +1,3270 @@
+# Copyright (c) 2005 Divmod, Inc.
+# Copyright (c) Twisted Matrix Laboratories.
+# See LICENSE.Twisted for details.
+
+"""
+Tests for L{provisioningserver.rpc.amp32}.
+"""
+
+from __future__ import (
+    absolute_import,
+    print_function,
+    # unicode_literals,
+    )
+
+str = None
+
+__metaclass__ = type
+__all__ = []
+
+import datetime
+import decimal
+from warnings import (
+    catch_warnings,
+    simplefilter,
+    )
+
+from maastesting.fixtures import TempDirectory
+from maastesting.testcase import (
+    MAASTestCase,
+    MAASTwistedRunTest,
+    )
+from provisioningserver.rpc import amp32
+from provisioningserver.rpc.testing import TwistedLoggerFixture
+from testtools.deferredruntest import assert_fails_with
+from testtools.matchers import (
+    Equals,
+    HasLength,
+    Is,
+    IsInstance,
+    StartsWith,
+    )
+from twisted.internet import (
+    defer,
+    error,
+    interfaces,
+    protocol,
+    reactor,
+    )
+from twisted.internet.address import UNIXAddress
+from twisted.internet.error import ConnectionLost
+from twisted.python import filepath
+from twisted.python.failure import Failure
+from twisted.test import iosim
+from twisted.test.proto_helpers import StringTransport
+from zope.interface import implements
+from zope.interface.verify import (
+    verifyClass,
+    verifyObject,
+    )
+
+
+try:
+    from twisted.internet import ssl
+except ImportError:
+    ssl = None
+else:
+    if not ssl.supported:
+        ssl = None
+
+if ssl is None:
+    skipSSL = "SSL not available"
+else:
+    skipSSL = None
+
+
+class AMP32TestCase(MAASTestCase):
+    """Common test class for testing AMP with 32-bit length prefixes.
+
+    All the tests in this module, as well as the implementation in `amp32`,
+    are derived from Twisted's `amp` implementation. Twisted uses Trial and
+    the AMP tests depend on it. Many of these dependencies have been recast
+    into uses of `testtools` or plain `unittest` and so on. This class remains
+    as the retirement home for the last few bits for which there are not yet
+    clean and uninvasive replacements.
+    """
+
+    run_tests_with = MAASTwistedRunTest.make_factory(timeout=5)
+
+    def setUp(self):
+        super(AMP32TestCase, self).setUp()
+        # Capture all Twisted logs. Normally this isn't necessary, but these
+        # tests are very noisy without this. This is also required for
+        # getLoggedFailures().
+        self.logger = self.useFixture(TwistedLoggerFixture())
+
+    def getLoggedFailures(self, error_type):
+        """Replacement for Trial's `flushLoggedErrors`.
+
+        This only queries the log for errors, whereas `flushLoggedErrors` also
+        removes those errors from the log. The reason for this is that Twisted
+        will fail a test if any errors are logged.
+
+        In time it would make sense to adopt Trial's approach. That's too
+        invasive at present, so this remains here as a reminder.
+
+        See :py:func:`twisted.trial.testcase.TestCase.flushLoggedErrors`.
+        """
+        errors = (log for log in self.logger.logs if log["isError"])
+        failures = (log['failure'] for log in errors if 'failure' in log)
+        return list(f for f in failures if f.check(error_type))
+
+    def assertWarns(self, category, message, filename, f, *args, **kwargs):
+        """Replacement for Trial's `assertWarns`.
+
+        Trial's implementation is slightly more featureful, but this does much
+        the same. Could be recast as a :py:class:`testtools.matchers.Matcher`.
+
+        See :py:func:`twisted.trial.testcase.TestCase.assertWarns`.
+        """
+        with catch_warnings(record=True) as log:
+            simplefilter('always')
+            result = f(*args, **kwargs)
+
+        self.assertThat(log, HasLength(1))
+        [warning] = log
+        self.expectThat(category, Is(warning.category))
+        self.expectThat(warning.message, IsInstance(Warning))
+        self.expectThat(message, Equals(warning.message[0]))
+        self.expectThat(filename, StartsWith(warning.filename))
+
+        return result
+
+
+class TestProto(protocol.Protocol):
+    """
+    A trivial protocol for use in testing where a L{Protocol} is expected.
+
+    @ivar instanceId: the id of this instance
+    @ivar onConnLost: deferred that will fired when the connection is lost
+    @ivar dataToSend: data to send on the protocol
+    """
+
+    instanceCount = 0
+
+    def __init__(self, onConnLost, dataToSend):
+        self.onConnLost = onConnLost
+        self.dataToSend = dataToSend
+        self.instanceId = TestProto.instanceCount
+        TestProto.instanceCount = TestProto.instanceCount + 1
+
+    def connectionMade(self):
+        self.data = []
+        self.transport.write(self.dataToSend)
+
+    def dataReceived(self, bytes):
+        self.data.append(bytes)
+
+    def connectionLost(self, reason):
+        self.onConnLost.callback(self.data)
+
+    def __repr__(self):
+        """
+        Custom repr for testing to avoid coupling amp tests with repr from
+        L{Protocol}
+
+        Returns a string which contains a unique identifier that can be looked
+        up using the instanceId property::
+
+            <TestProto #3>
+        """
+        return "<TestProto #%d>" % (self.instanceId,)
+
+
+class SimpleSymmetricProtocol(amp32.AMP):
+
+    def sendHello(self, text):
+        return self.callRemoteString("hello", hello=text)
+
+    def amp_HELLO(self, box):
+        return amp32.Box(hello=box['hello'])
+
+    def amp_HOWDOYOUDO(self, box):
+        return amp32.QuitBox(howdoyoudo='world')
+
+
+class UnfriendlyGreeting(Exception):
+    """Greeting was insufficiently kind.
+    """
+
+
+class DeathThreat(Exception):
+    """Greeting was insufficiently kind.
+    """
+
+
+class UnknownProtocol(Exception):
+    """Asked to switch to the wrong protocol.
+    """
+
+
+class TransportPeer(amp32.Argument):
+
+    # this serves as some informal documentation for how to get variables from
+    # the protocol or your environment and pass them to methods as arguments.
+    def retrieve(self, d, name, proto):
+        return ''
+
+    def fromStringProto(self, notAString, proto):
+        return proto.transport.getPeer()
+
+    def toBox(self, name, strings, objects, proto):
+        return
+
+
+class Hello(amp32.Command):
+
+    commandName = 'hello'
+
+    arguments = [
+        ('hello', amp32.String()),
+        ('optional', amp32.Boolean(optional=True)),
+        ('print', amp32.Unicode(optional=True)),
+        ('from', TransportPeer(optional=True)),
+        ('mixedCase', amp32.String(optional=True)),
+        ('dash-arg', amp32.String(optional=True)),
+        ('underscore_arg', amp32.String(optional=True)),
+    ]
+
+    response = [
+        ('hello', amp32.String()),
+        ('print', amp32.Unicode(optional=True)),
+    ]
+
+    errors = {UnfriendlyGreeting: 'UNFRIENDLY'}
+
+    fatalErrors = {DeathThreat: 'DEAD'}
+
+
+class NoAnswerHello(Hello):
+
+    commandName = Hello.commandName
+    requiresAnswer = False
+
+
+class FutureHello(amp32.Command):
+
+    commandName = 'hello'
+    arguments = [
+        ('hello', amp32.String()),
+        ('optional', amp32.Boolean(optional=True)),
+        ('print', amp32.Unicode(optional=True)),
+        ('from', TransportPeer(optional=True)),
+        # addt'l arguments should generally be added at the end, and be
+        # optional...
+        ('bonus', amp32.String(optional=True)),
+    ]
+
+    response = [
+        ('hello', amp32.String()),
+        ('print', amp32.Unicode(optional=True)),
+    ]
+
+    errors = {UnfriendlyGreeting: 'UNFRIENDLY'}
+
+
+class WTF(amp32.Command):
+    """
+    An example of an invalid command.
+    """
+
+
+class BrokenReturn(amp32.Command):
+    """
+    An example of a perfectly good command, but the handler is going to return
+    None...
+    """
+
+    commandName = 'broken_return'
+
+
+class Goodbye(amp32.Command):
+
+    # commandName left blank on purpose: this tests implicit command names.
+    response = [('goodbye', amp32.String())]
+    responseType = amp32.QuitBox
+
+
+class Howdoyoudo(amp32.Command):
+
+    commandName = 'howdoyoudo'
+    # responseType = amp32.QuitBox
+
+
+class WaitForever(amp32.Command):
+
+    commandName = 'wait_forever'
+
+
+class GetList(amp32.Command):
+
+    commandName = 'getlist'
+    arguments = [('length', amp32.Integer())]
+    response = [('body', amp32.AmpList([('x', amp32.Integer())]))]
+
+
+class DontRejectMe(amp32.Command):
+
+    commandName = 'dontrejectme'
+    arguments = [
+        ('magicWord', amp32.Unicode()),
+        ('list', amp32.AmpList([('name', amp32.Unicode())], optional=True)),
+    ]
+    response = [('response', amp32.Unicode())]
+
+
+class SecuredPing(amp32.Command):
+
+    # XXX TODO: actually make this refuse to send over an insecure connection
+    response = [('pinged', amp32.Boolean())]
+
+
+class TestSwitchProto(amp32.ProtocolSwitchCommand):
+
+    commandName = 'Switch-Proto'
+    arguments = [
+        ('name', amp32.String()),
+        ]
+    errors = {UnknownProtocol: 'UNKNOWN'}
+
+
+class SingleUseFactory(protocol.ClientFactory):
+
+    def __init__(self, proto):
+        self.proto = proto
+        self.proto.factory = self
+
+    def buildProtocol(self, addr):
+        p, self.proto = self.proto, None
+        return p
+
+    reasonFailed = None
+
+    def clientConnectionFailed(self, connector, reason):
+        self.reasonFailed = reason
+        return
+
+
+THING_I_DONT_UNDERSTAND = 'gwebol nargo'
+
+
+class ThingIDontUnderstandError(Exception):
+    pass
+
+
+class FactoryNotifier(amp32.AMP):
+
+    factory = None
+
+    def connectionMade(self):
+        if self.factory is not None:
+            self.factory.theProto = self
+            if hasattr(self.factory, 'onMade'):
+                self.factory.onMade.callback(None)
+
+    @SecuredPing.responder
+    def emitpong(self):
+        from twisted.internet.interfaces import ISSLTransport
+        if not ISSLTransport.providedBy(self.transport):
+            raise DeathThreat("only send secure pings over secure channels")
+        return {'pinged': True}
+
+
+class SimpleSymmetricCommandProtocol(FactoryNotifier):
+
+    maybeLater = None
+
+    def __init__(self, onConnLost=None):
+        amp32.AMP.__init__(self)
+        self.onConnLost = onConnLost
+
+    def sendHello(self, text):
+        return self.callRemote(Hello, hello=text)
+
+    def sendUnicodeHello(self, text, translation):
+        return self.callRemote(Hello, hello=text, Print=translation)
+
+    greeted = False
+
+    @Hello.responder
+    def cmdHello(self, hello, From, optional=None, Print=None,
+                 mixedCase=None, dash_arg=None, underscore_arg=None):
+        assert From == self.transport.getPeer()
+        if hello == THING_I_DONT_UNDERSTAND:
+            raise ThingIDontUnderstandError()
+        if hello.startswith('fuck'):
+            raise UnfriendlyGreeting("Don't be a dick.")
+        if hello == 'die':
+            raise DeathThreat("aieeeeeeeee")
+        result = dict(hello=hello)
+        if Print is not None:
+            result.update(dict(Print=Print))
+        self.greeted = True
+        return result
+
+    @GetList.responder
+    def cmdGetlist(self, length):
+        return {'body': [dict(x=1)] * length}
+
+    @DontRejectMe.responder
+    def okiwont(self, magicWord, list=None):
+        if list is None:
+            response = u'list omitted'
+        else:
+            response = u'%s accepted' % (list[0]['name'])
+        return dict(response=response)
+
+    @WaitForever.responder
+    def waitforit(self):
+        self.waiting = defer.Deferred()
+        return self.waiting
+
+    @Howdoyoudo.responder
+    def howdo(self):
+        return dict(howdoyoudo='world')
+
+    @Goodbye.responder
+    def saybye(self):
+        return dict(goodbye="everyone")
+
+    def switchToTestProtocol(self, fail=False):
+        if fail:
+            name = 'no-proto'
+        else:
+            name = 'test-proto'
+        p = TestProto(self.onConnLost, SWITCH_CLIENT_DATA)
+        return self.callRemote(
+            TestSwitchProto,
+            SingleUseFactory(p), name=name).addCallback(lambda ign: p)
+
+    @TestSwitchProto.responder
+    def switchit(self, name):
+        if name == 'test-proto':
+            return TestProto(self.onConnLost, SWITCH_SERVER_DATA)
+        raise UnknownProtocol(name)
+
+    @BrokenReturn.responder
+    def donothing(self):
+        return None
+
+
+class DeferredSymmetricCommandProtocol(SimpleSymmetricCommandProtocol):
+
+    @TestSwitchProto.responder
+    def switchit(self, name):
+        if name == 'test-proto':
+            self.maybeLaterProto = TestProto(
+                self.onConnLost, SWITCH_SERVER_DATA)
+            self.maybeLater = defer.Deferred()
+            return self.maybeLater
+        raise UnknownProtocol(name)
+
+
+class BadNoAnswerCommandProtocol(SimpleSymmetricCommandProtocol):
+
+    @NoAnswerHello.responder
+    def badResponder(
+            self, hello, From, optional=None, Print=None, mixedCase=None,
+            dash_arg=None, underscore_arg=None):
+        """
+        This responder does nothing and forgets to return a dictionary.
+        """
+
+
+class NoAnswerCommandProtocol(SimpleSymmetricCommandProtocol):
+
+    @NoAnswerHello.responder
+    def goodNoAnswerResponder(
+            self, hello, From, optional=None, Print=None,
+            mixedCase=None, dash_arg=None, underscore_arg=None):
+        return dict(hello=hello + "-noanswer")
+
+
+def connectedServerAndClient(
+        ServerClass=SimpleSymmetricProtocol,
+        ClientClass=SimpleSymmetricProtocol,
+        *a, **kw):
+    """Returns a 3-tuple: (client, server, pump)
+    """
+    return iosim.connectedServerAndClient(
+        ServerClass, ClientClass,
+        *a, **kw)
+
+
+class TotallyDumbProtocol(protocol.Protocol):
+
+    buf = ''
+
+    def dataReceived(self, data):
+        self.buf += data
+
+
+class LiteralAmp(amp32.AMP):
+
+    def __init__(self):
+        self.boxes = []
+
+    def ampBoxReceived(self, box):
+        self.boxes.append(box)
+        return
+
+
+class AmpBoxTests(AMP32TestCase):
+    """
+    Test a few essential properties of AMP boxes, mostly with respect to
+    serialization correctness.
+    """
+
+    def test_serializeStr(self):
+        """
+        Make sure that strs serialize to strs.
+        """
+        a = amp32.AmpBox(key='value')
+        self.assertEqual(type(a.serialize()), bytes)
+
+    def test_serializeUnicodeKeyRaises(self):
+        """
+        Verify that TypeError is raised when trying to serialize Unicode keys.
+        """
+        a = amp32.AmpBox(**{u'key': 'value'})
+        self.assertRaises(TypeError, a.serialize)
+
+    def test_serializeUnicodeValueRaises(self):
+        """
+        Verify that TypeError is raised when trying to serialize Unicode
+        values.
+        """
+        a = amp32.AmpBox(key=u'value')
+        self.assertRaises(TypeError, a.serialize)
+
+
+class ParsingTest(AMP32TestCase):
+
+    def test_booleanValues(self):
+        """
+        Verify that the Boolean parser parses 'True' and 'False', but nothing
+        else.
+        """
+        b = amp32.Boolean()
+        self.assertEqual(b.fromString("True"), True)
+        self.assertEqual(b.fromString("False"), False)
+        self.assertRaises(TypeError, b.fromString, "ninja")
+        self.assertRaises(TypeError, b.fromString, "true")
+        self.assertRaises(TypeError, b.fromString, "TRUE")
+        self.assertEqual(b.toString(True), 'True')
+        self.assertEqual(b.toString(False), 'False')
+
+    def test_pathValueRoundTrip(self):
+        """
+        Verify the 'Path' argument can parse and emit a file path.
+        """
+        with TempDirectory() as tempdir:
+            fp = filepath.FilePath(tempdir.path)
+            p = amp32.Path()
+            s = p.toString(fp)
+            v = p.fromString(s)
+            self.assertIsNot(fp, v)  # sanity check
+            self.assertEqual(fp, v)
+
+    def test_sillyEmptyThing(self):
+        """
+        Test that empty boxes raise an error; they aren't supposed to be sent
+        on purpose.
+        """
+        a = amp32.AMP()
+        return self.assertRaises(
+            amp32.NoEmptyBoxes, a.ampBoxReceived, amp32.Box())
+
+    def test_ParsingRoundTrip(self):
+        """
+        Verify that various kinds of data make it through the encode/parse
+        round-trip unharmed.
+        """
+        c, s, p = connectedServerAndClient(
+            ClientClass=LiteralAmp, ServerClass=LiteralAmp)
+
+        SIMPLE = ('simple', 'test')
+        CE = ('ceq', ': ')
+        CR = ('crtest', 'test\r')
+        LF = ('lftest', 'hello\n')
+        NEWLINE = ('newline', 'test\r\none\r\ntwo')
+        NEWLINE2 = ('newline2', 'test\r\none\r\n two')
+        BODYTEST = ('body', 'blah\r\n\r\ntesttest')
+
+        testData = [
+            [SIMPLE],
+            [SIMPLE, BODYTEST],
+            [SIMPLE, CE],
+            [SIMPLE, CR],
+            [SIMPLE, CE, CR, LF],
+            [CE, CR, LF],
+            [SIMPLE, NEWLINE, CE, NEWLINE2],
+            [BODYTEST, SIMPLE, NEWLINE]
+        ]
+
+        for test in testData:
+            jb = amp32.Box()
+            jb.update(dict(test))
+            jb._sendTo(c)
+            p.flush()
+            self.assertEqual(s.boxes[-1], jb)
+
+
+class FakeLocator(object):
+    """
+    This is a fake implementation of the interface implied by
+    L{CommandLocator}.
+    """
+
+    def __init__(self):
+        """
+        Remember the given keyword arguments as a set of responders.
+        """
+        self.commands = {}
+
+    def locateResponder(self, commandName):
+        """
+        Look up and return a function passed as a keyword argument of the given
+        name to the constructor.
+        """
+        return self.commands[commandName]
+
+
+class FakeSender:
+    """
+    This is a fake implementation of the 'box sender' interface implied by
+    L{AMP}.
+    """
+
+    def __init__(self):
+        """
+        Create a fake sender and initialize the list of received boxes and
+        unhandled errors.
+        """
+        self.sentBoxes = []
+        self.unhandledErrors = []
+        self.expectedErrors = 0
+
+    def expectError(self):
+        """
+        Expect one error, so that the test doesn't fail.
+        """
+        self.expectedErrors += 1
+
+    def sendBox(self, box):
+        """
+        Accept a box, but don't do anything.
+        """
+        self.sentBoxes.append(box)
+
+    def unhandledError(self, failure):
+        """
+        Deal with failures by instantly re-raising them for easier debugging.
+        """
+        self.expectedErrors -= 1
+        if self.expectedErrors < 0:
+            failure.raiseException()
+        else:
+            self.unhandledErrors.append(failure)
+
+
+class CommandDispatchTests(AMP32TestCase):
+    """
+    The AMP CommandDispatcher class dispatches converts AMP boxes into commands
+    and responses using Command.responder decorator.
+