Add functional tests for device auth

CHANGES.rst

@@ -6,16 +6,23 @@ CHANGES
 
 Changes:
 
-- Fix atvremote exit codes
+- Support AirPlay device authentication
 - Support arrow keys (left, right, up, down)
 - Support scanning for Apple TVs with home sharing disabled
 - Support for shuffle and repeat modes
 - Support for "stop" button
 - Multiple commands can be given to atvremote
 - Handle additional media kinds
+- Fix atvremote exit codes
 - Support python 3.6
 - Bump aiohttp to 1.3.5 and support 2.0.0+
 
+Notes:
+
+- play_url has moved to the new airplay module and no longer
+  accepts start position as required argument. This is a
+  breaking change!
+
 Other:
 
 - Upgrade test tools (pylint, flake, etc.)

README.rst

@@ -75,7 +75,7 @@ It is possible to use the reference CLI application as well:
     # Scanning for devices on network
     $ atvremote
     Found Apple TVs:
-     - Apple TV at 10.0.10.22 (hsgid: 00000000-1234-5678-9012- 345678901234)
+     - Apple TV at 10.0.10.22 (hsgid: 00000000-1234-5678-9012-345678901234)
 
     Note: You must use 'pair' with devices that have home sharing disabled
 
@@ -85,6 +85,8 @@ It is possible to use the reference CLI application as well:
     Play state: Playing
       Position: 0/397s (0.0%)
 
+    # Passing multiple commands
+    $ atvremote -a next next play playing stop
 
     # List all commands supported by a device
     $ atvremote -a commands

pyatv/__init__.py

@@ -20,13 +20,15 @@
 
 
 class AppleTVDevice(
-        namedtuple('AppleTVDevice', 'name, address, login_id, port')):
+        namedtuple('AppleTVDevice',
+                   'name, address, login_id, port, airplay_port')):
     """Representation of an Apple TV device used when connecting."""
 
-    def __new__(cls, name, address, login_id, port=3689):
+    # pylint: disable=too-many-arguments
+    def __new__(cls, name, address, login_id, port=3689, airplay_port=7000):
         """Initialize a new AppleTVDevice."""
         return super(AppleTVDevice, cls).__new__(
-            cls, name, address, login_id, port)
+            cls, name, address, login_id, port, airplay_port)
 
 
 # pylint: disable=too-few-public-methods

pyatv/internal/apple_tv.py

@@ -488,7 +488,8 @@ def __init__(self, loop, session, details):
 
         airplay_player = player.AirPlayPlayer(loop, session, details.address)
         airplay_http = HttpSession(
-            session, 'http://{0}:7000/'.format(details.address))
+            session, 'http://{0}:{1}/'.format(
+                details.address, details.airplay_port))
         self._airplay = AirPlayInternal(airplay_http, airplay_player)
 
     def login(self):

tests/fake_apple_tv.py

@@ -31,9 +31,10 @@
 # --- START AUTHENTICATION DATA VALID SESSION (FROM DEVICE) ---
 
 DEVICE_IDENTIFIER = '75FBEEC773CFC563'
-DEVICE_AUTH_KEY = binascii.unhexlify(
-    '8F06696F2542D70DF59286C761695C485F815BE3D152849E1361282D46AB1493')
+DEVICE_AUTH_KEY = \
+    '8F06696F2542D70DF59286C761695C485F815BE3D152849E1361282D46AB1493'
 DEVICE_PIN = 2271
+DEVICE_CREDENTIALS = DEVICE_IDENTIFIER + ':' + DEVICE_AUTH_KEY
 
 # pylint: disable=E501
 # For authentication
@@ -100,6 +101,8 @@ def __init__(self, loop, hsgid, pairing_guid, session_id, testcase):
         self.session = None
         self.last_button_pressed = None
         self.buttons_press_count = 0
+        self.has_authenticated = True
+        self.last_airplay_url = None
         self.properties = {}  # setproperty
         self.tc = testcase
 
@@ -331,6 +334,9 @@ def _verify_auth_parameters(self,
     @asyncio.coroutine
     def handle_airplay_play(self, request):
         """Handle AirPlay play requests."""
+        if not self.has_authenticated:
+            return web.Response(status=503)
+
         headers = request.headers
 
         # Verify headers first
@@ -385,6 +391,7 @@ def handle_airplay_pair_verify(self, request):
             return web.Response(
                 body=binascii.unhexlify(_DEVICE_VERIFY_STEP1_RESP), status=200)
         elif hexlified == _DEVICE_VERIFY_STEP2:
+            self.has_authenticated = True
             return web.Response(body=_DEVICE_VERIFY_STEP2_RESP, status=200)
 
         return web.Response(body=b'', status=503)
@@ -488,6 +495,10 @@ def airplay_playback_playing(self):
         self.device.responses['airplay_playback'].insert(
             0, AirPlayPlaybackResponse(plistlib.dumps(plist)))
 
+    def airplay_require_authentication(self):
+        """Require device authentication for AirPlay."""
+        self.device.has_authenticated = False
+
     def pairing_response(self, remote_name, expected_pairing_code):
         """Reponse when a pairing request is made."""
         self.device.responses['pairing'].insert(0, PairingResponse(

tests/test_airplay_auth.py

@@ -58,7 +58,7 @@ def test_verify_authenticated(self):
         http = HttpSession(
             self.session, 'http://127.0.0.1:{0}/'.format(self.app.port))
         handler = srp.SRPAuthHandler()
-        handler.initialize(DEVICE_AUTH_KEY)
+        handler.initialize(binascii.unhexlify(DEVICE_AUTH_KEY))
 
         verifier = AuthenticationVerifier(http, handler)
         self.assertTrue((yield from verifier.verify_authed()))
@@ -81,7 +81,7 @@ def test_auth_failed(self):
         http = HttpSession(
             self.session, 'http://127.0.0.1:{0}/'.format(self.app.port))
         handler = srp.SRPAuthHandler()
-        handler.initialize(DEVICE_AUTH_KEY)
+        handler.initialize(binascii.unhexlify(DEVICE_AUTH_KEY))
 
         auther = DeviceAuthenticator(http, handler)
         yield from auther.start_authentication()

tests/test_functional.py

@@ -9,7 +9,8 @@
 
 from pyatv import (AppleTVDevice, connect_to_apple_tv, const,
                    exceptions, pairing)
-from tests.fake_apple_tv import (FakeAppleTV, AppleTVUseCases)
+from tests.fake_apple_tv import (
+    FakeAppleTV, AppleTVUseCases, DEVICE_PIN, DEVICE_CREDENTIALS)
 from tests import (utils, zeroconf_stub)
 
 HSGID = '12345-6789-0'
@@ -70,7 +71,7 @@ def get_application(self, loop=None):
 
     def get_connected_device(self, identifier):
         details = AppleTVDevice(
-            'Apple TV', '127.0.0.1', identifier, self.app.port)
+            'Apple TV', '127.0.0.1', identifier, self.app.port, self.app.port)
         return connect_to_apple_tv(details, self.loop)
 
     # This is not a pretty test and it does crazy things. Should probably be
@@ -89,6 +90,18 @@ def test_pairing_with_device(self):
 
         self.assertTrue(handler.has_paired, msg='did not pair with device')
 
+    @unittest_run_loop
+    def test_device_authentication(self):
+        # Credentials used for device authentication
+        yield from self.atv.airplay.load_credentials(DEVICE_CREDENTIALS)
+
+        # Perform authentication
+        yield from self.atv.airplay.start_authentication()
+        yield from self.atv.airplay.finish_authentication(DEVICE_PIN)
+
+        # Verify credentials are authenticated
+        self.assertTrue((yield from self.atv.airplay.verify_authenticated()))
+
     @unittest_run_loop
     def test_play_url(self):
         self.usecase.airplay_playback_idle()
@@ -100,6 +113,20 @@ def test_play_url(self):
 
         self.assertEqual(self.fake_atv.last_airplay_url, AIRPLAY_STREAM)
 
+    @unittest_run_loop
+    def test_play_url_authenticated(self):
+        self.usecase.airplay_require_authentication()
+        self.usecase.airplay_playback_idle()
+        self.usecase.airplay_playback_playing()
+        self.usecase.airplay_playback_idle()
+
+        yield from self.atv.airplay.load_credentials(DEVICE_CREDENTIALS)
+
+        yield from self.atv.airplay.play_url(
+            AIRPLAY_STREAM, port=self.app.port)
+
+        self.assertEqual(self.fake_atv.last_airplay_url, AIRPLAY_STREAM)
+
     @unittest_run_loop
     def test_login_failed(self):
         # Twice since the client will retry one time