Merge pull request #33 from desec-io/20170210_basicAuth

fix(api): fix HTTP error codes for basic auth, closes #26

api/desecapi/authentication.py

@@ -4,6 +4,7 @@ import base64
 from rest_framework import exceptions, HTTP_HEADER_ENCODING
 from rest_framework.authtoken.models import Token
 from rest_framework.authentication import BaseAuthentication, get_authorization_header, authenticate
+from desecapi.models import Domain
 
 
 class BasicTokenAuthentication(BaseAuthentication):
@@ -52,6 +53,12 @@ class BasicTokenAuthentication(BaseAuthentication):
         if not token.user.is_active:
             raise exceptions.AuthenticationFailed('User inactive or deleted')
 
+        if user:
+            try:
+                Domain.objects.get(owner=token.user.pk, name=user)
+            except:
+                raise exceptions.AuthenticationFailed('Invalid username')
+
         return token.user, token
 
     def authenticate_header(self, request):

api/desecapi/tests/testdyndns12update.py

@@ -130,6 +130,32 @@ class DynDNS12UpdateTest(APITestCase):
         self.assertEqual(response.data, 'good')
         self.assertIP(ipv4='10.5.5.5')
 
+        self.client.credentials(HTTP_AUTHORIZATION='Basic ' + base64.b64encode((self.username + '.invalid:' + self.password).encode()).decode())
+        url = reverse('dyndns12update')
+        response = self.client.get(url, REMOTE_ADDR='10.5.5.5')
+        self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)
+
+    def testIdentificationByTokenWithEmptyUser(self):
+        self.client.credentials(HTTP_AUTHORIZATION='Basic ' + base64.b64encode((':' + self.password).encode()).decode())
+        url = reverse('dyndns12update')
+        response = self.client.get(url, REMOTE_ADDR='10.5.5.6')
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        self.assertEqual(response.data, 'good')
+        self.assertIP(ipv4='10.5.5.6')
+
+        # Now make sure we get a conflict when the user has multiple domains. Thus,
+        # we add a second domain for the current user.
+
+        self.client.credentials(HTTP_AUTHORIZATION='Token ' + self.token)
+        url = reverse('domain-list')
+        data = {'name': 'second-' + self.domain}
+        response = self.client.post(url, data)
+        self.assertEqual(response.status_code, status.HTTP_201_CREATED)
+
+        url = reverse('dyndns12update')
+        response = self.client.get(url, REMOTE_ADDR='10.5.5.7')
+        self.assertEqual(response.status_code, status.HTTP_409_CONFLICT)
+
     def testManualIPv6(self):
         #/update?username=foobar.dedyn.io&password=secret
         self.client.credentials(HTTP_AUTHORIZATION='')

api/desecapi/views.py

@@ -171,7 +171,9 @@ class DynDNS12Update(APIView):
 
             # 3. http basic auth username
             try:
-                return base64.b64decode(get_authorization_header(request).decode().split(' ')[1].encode()).decode().split(':')[0]
+                domainname = base64.b64decode(get_authorization_header(request).decode().split(' ')[1].encode()).decode().split(':')[0]
+                if domainname:
+                    return domainname
             except IndexError:
                 pass
             except UnicodeDecodeError:
@@ -184,6 +186,10 @@ class DynDNS12Update(APIView):
             # 5. only domain associated with this user account
             if len(request.user.domains.all()) == 1:
                 return request.user.domains.all()[0].name
+            if len(request.user.domains.all()) > 1:
+                ex = ValidationError(detail={"detail": "Request does not specify domain unambiguously.", "code": "domain-ambiguous"})
+                ex.status_code = status.HTTP_409_CONFLICT
+                raise ex
 
             return None