Add an autogenerated http_user field to the host table.

.gitignore

@@ -1,3 +1,4 @@
+.venv
 *.log
 *.pot
 *.pyc

.vscode/settings.json

@@ -0,0 +1,3 @@
+{
+    "python.pythonPath": ".venv\\Scripts\\python.exe"
+}

CHANGES.rst

@@ -7,6 +7,11 @@ AFTER that, you can upgrade to 0.10.x or any later version (and then run the
 migrations for that version). For upgrading and migration help, please see
 the docs that match the version you are upgrading to.
 
+Release 0.13.0-alternative
+---------------------------------
+
+New Features:
+ - use a http user different from the fqdn of the host due to limitation in the length of the username from some routers
 
 Release 0.13.0 (not released yet)
 ---------------------------------

conftest.py

@@ -7,6 +7,7 @@
 from random import randint
 
 from nsupdate.main.dnstools import FQDN
+from nsupdate.main.models import _HOST_HTTP_USER_MAX_LENGTH
 
 from django.conf import settings
 
@@ -40,6 +41,10 @@
 from nsupdate.main.dnstools import update_ns, FQDN
 
 
+def to_http_user(hostname):
+    return str(hostname)[:_HOST_HTTP_USER_MAX_LENGTH].zfill(_HOST_HTTP_USER_MAX_LENGTH)
+
+
 @pytest.fixture(scope="function")
 def ddns_hostname():
     """
@@ -96,10 +101,12 @@ def db_init(db):  # note: db is a predefined fixture and required here to have t
     )
     # a Host for api / session update tests
     hostname = TEST_HOST.host
-    h = Host(name=hostname, domain=dt, created_by=u, netmask_ipv4=29, netmask_ipv6=64)
+    h = Host(name=hostname, domain=dt, created_by=u, netmask_ipv4=29, netmask_ipv6=64,
+             http_user=to_http_user(TEST_HOST))
     h.generate_secret(secret=TEST_SECRET)
     hostname2 = TEST_HOST2.host
-    h2 = Host(name=hostname2, domain=dt, created_by=u2, netmask_ipv4=29, netmask_ipv6=64)
+    h2 = Host(name=hostname2, domain=dt, created_by=u2, netmask_ipv4=29, netmask_ipv6=64,
+              http_user=to_http_user(TEST_HOST2))
     h2.generate_secret(secret=TEST_SECRET2)
 
     # "update other service" ddns_client feature

src/nsupdate/api/_tests/test_api.py

@@ -13,7 +13,7 @@
 from nsupdate.main.models import Domain
 from nsupdate.api.views import basic_authenticate
 
-from conftest import TESTDOMAIN, TEST_HOST, TEST_HOST_RELATED, TEST_HOST2, TEST_SECRET, TEST_SECRET2
+from conftest import TESTDOMAIN, TEST_HOST, TEST_HOST_RELATED, TEST_HOST2, TEST_SECRET, TEST_SECRET2, to_http_user
 
 USERNAME = 'test'
 PASSWORD = 'pass'
@@ -58,58 +58,51 @@ def test_basic_auth():
 
 def test_nic_update_badauth(client):
     response = client.get(reverse('nic_update'),
-                          HTTP_AUTHORIZATION=make_basic_auth_header(TEST_HOST, "wrong"))
+                          HTTP_AUTHORIZATION=make_basic_auth_header(to_http_user(TEST_HOST), "wrong"))
     assert response.status_code == 401
     assert response.content == b'badauth'
 
 
 def test_nic_update_authorized_nonexistent_host(client):
     response = client.get(reverse('nic_update') + '?hostname=nonexistent.nsupdate.info',
-                          HTTP_AUTHORIZATION=make_basic_auth_header(TEST_HOST, TEST_SECRET))
+                          HTTP_AUTHORIZATION=make_basic_auth_header(to_http_user(TEST_HOST), TEST_SECRET))
     assert response.status_code == 200
     # we must not get this updated, it doesn't exist in the database:
     assert response.content == b'nohost'
 
 
 def test_nic_update_authorized_foreign_host(client):
     response = client.get(reverse('nic_update') + '?hostname=%s' % (TEST_HOST2, ),
-                          HTTP_AUTHORIZATION=make_basic_auth_header(TEST_HOST, TEST_SECRET))
+                          HTTP_AUTHORIZATION=make_basic_auth_header(to_http_user(TEST_HOST), TEST_SECRET))
     assert response.status_code == 200
     # we must not get this updated, this is a host of some other user!
     assert response.content == b'nohost'
 
 
 def test_nic_update_authorized_not_fqdn_hostname(client):
     response = client.get(reverse('nic_update') + '?hostname=test',
-                          HTTP_AUTHORIZATION=make_basic_auth_header(TEST_HOST, TEST_SECRET))
-    assert response.status_code == 200
-    assert response.content == b'notfqdn'
-
-
-def test_nic_update_authorized_not_fqdn_username(client):
-    response = client.get(reverse('nic_update'),
-                          HTTP_AUTHORIZATION=make_basic_auth_header('test', TEST_SECRET))
+                          HTTP_AUTHORIZATION=make_basic_auth_header(to_http_user(TEST_HOST), TEST_SECRET))
     assert response.status_code == 200
     assert response.content == b'notfqdn'
 
 
 def test_nic_update_authorized_invalid_ip1(client):
     response = client.get(reverse('nic_update') + '?myip=1234',
-                          HTTP_AUTHORIZATION=make_basic_auth_header(TEST_HOST, TEST_SECRET))
+                          HTTP_AUTHORIZATION=make_basic_auth_header(to_http_user(TEST_HOST), TEST_SECRET))
     assert response.status_code == 200
     assert response.content == b'dnserr'
 
 
 def test_nic_update_authorized_invalid_ip2(client):
     response = client.get(reverse('nic_update') + '?myip=%C3%A4%C3%BC%C3%B6',
-                          HTTP_AUTHORIZATION=make_basic_auth_header(TEST_HOST, TEST_SECRET))
+                          HTTP_AUTHORIZATION=make_basic_auth_header(to_http_user(TEST_HOST), TEST_SECRET))
     assert response.status_code == 200
     assert response.content == b'dnserr'
 
 
 def test_nic_update_authorized(client):
     response = client.get(reverse('nic_update'),
-                          HTTP_AUTHORIZATION=make_basic_auth_header(TEST_HOST, TEST_SECRET))
+                          HTTP_AUTHORIZATION=make_basic_auth_header(to_http_user(TEST_HOST), TEST_SECRET))
     assert response.status_code == 200
     # we don't care whether it is nochg or good, but should be one of them:
     content = response.content.decode('utf-8')
@@ -122,28 +115,28 @@ def test_nic_update_authorized_ns_unavailable(client):
     d.save()
     # prepare: we must make sure the real test is not a nochg update
     response = client.get(reverse('nic_update') + '?myip=1.2.3.4',
-                          HTTP_AUTHORIZATION=make_basic_auth_header(TEST_HOST, TEST_SECRET))
+                          HTTP_AUTHORIZATION=make_basic_auth_header(to_http_user(TEST_HOST), TEST_SECRET))
     assert response.status_code == 200
     # now do the real test: ip changed, but we can't update DNS as it is unavailable
     response = client.get(reverse('nic_update') + '?myip=4.3.2.1',
-                          HTTP_AUTHORIZATION=make_basic_auth_header(TEST_HOST, TEST_SECRET))
+                          HTTP_AUTHORIZATION=make_basic_auth_header(to_http_user(TEST_HOST), TEST_SECRET))
     assert response.status_code == 200
     assert response.content == b'dnserr'
 
 
 def test_nic_update_authorized_myip_v4(client):
     response = client.get(reverse('nic_update') + '?myip=4.3.2.1',
-                          HTTP_AUTHORIZATION=make_basic_auth_header(TEST_HOST, TEST_SECRET))
+                          HTTP_AUTHORIZATION=make_basic_auth_header(to_http_user(TEST_HOST), TEST_SECRET))
     assert response.status_code == 200
     # we don't care whether it is nochg or good, but should be the ip from myip=...:
     assert response.content in [b'good 4.3.2.1', b'nochg 4.3.2.1']
     response = client.get(reverse('nic_update') + '?myip=1.2.3.4',
-                          HTTP_AUTHORIZATION=make_basic_auth_header(TEST_HOST, TEST_SECRET))
+                          HTTP_AUTHORIZATION=make_basic_auth_header(to_http_user(TEST_HOST), TEST_SECRET))
     assert response.status_code == 200
     # must be good (was different IP)
     assert response.content == b'good 1.2.3.4'
     response = client.get(reverse('nic_update') + '?myip=1.2.3.4',
-                          HTTP_AUTHORIZATION=make_basic_auth_header(TEST_HOST, TEST_SECRET))
+                          HTTP_AUTHORIZATION=make_basic_auth_header(to_http_user(TEST_HOST), TEST_SECRET))
     assert response.status_code == 200
     # must be nochg (was same IP)
     assert response.content == b'nochg 1.2.3.4'
@@ -153,17 +146,17 @@ def test_nic_update_authorized_myip_v4(client):
 
 def test_nic_update_authorized_myip_v6(client):
     response = client.get(reverse('nic_update') + '?myip=2000::2',
-                          HTTP_AUTHORIZATION=make_basic_auth_header(TEST_HOST, TEST_SECRET))
+                          HTTP_AUTHORIZATION=make_basic_auth_header(to_http_user(TEST_HOST), TEST_SECRET))
     assert response.status_code == 200
     # we don't care whether it is nochg or good, but should be the ip from myip=...:
     assert response.content in [b'good 2000::2', b'nochg 2000::2']
     response = client.get(reverse('nic_update') + '?myip=2000::3',
-                          HTTP_AUTHORIZATION=make_basic_auth_header(TEST_HOST, TEST_SECRET))
+                          HTTP_AUTHORIZATION=make_basic_auth_header(to_http_user(TEST_HOST), TEST_SECRET))
     assert response.status_code == 200
     # must be good (was different IP)
     assert response.content == b'good 2000::3'
     response = client.get(reverse('nic_update') + '?myip=2000::3',
-                          HTTP_AUTHORIZATION=make_basic_auth_header(TEST_HOST, TEST_SECRET))
+                          HTTP_AUTHORIZATION=make_basic_auth_header(to_http_user(TEST_HOST), TEST_SECRET))
     assert response.status_code == 200
     # must be nochg (was same IP)
     assert response.content == b'nochg 2000::3'
@@ -174,12 +167,12 @@ def test_nic_update_authorized_myip_v6(client):
 @pytest.mark.requires_sequential
 def test_nic_update_authorized_update_other_services(client):
     response = client.get(reverse('nic_update') + '?myip=4.3.2.1',
-                          HTTP_AUTHORIZATION=make_basic_auth_header(TEST_HOST, TEST_SECRET))
+                          HTTP_AUTHORIZATION=make_basic_auth_header(to_http_user(TEST_HOST), TEST_SECRET))
     assert response.status_code == 200
     # we don't care whether it is nochg or good, but should be the ip from myip=...:
     assert response.content in [b'good 4.3.2.1', b'nochg 4.3.2.1']
     response = client.get(reverse('nic_update') + '?myip=1.2.3.4',
-                          HTTP_AUTHORIZATION=make_basic_auth_header(TEST_HOST, TEST_SECRET))
+                          HTTP_AUTHORIZATION=make_basic_auth_header(to_http_user(TEST_HOST), TEST_SECRET))
     assert response.status_code == 200
     # must be good (was different IP)
     assert response.content == b'good 1.2.3.4'
@@ -188,7 +181,7 @@ def test_nic_update_authorized_update_other_services(client):
     # now check if it updated the other service also:
     assert query_ns(TEST_HOST_OTHER, 'A') == '1.2.3.4'
     response = client.get(reverse('nic_update') + '?myip=2.3.4.5',
-                          HTTP_AUTHORIZATION=make_basic_auth_header(TEST_HOST, TEST_SECRET))
+                          HTTP_AUTHORIZATION=make_basic_auth_header(to_http_user(TEST_HOST), TEST_SECRET))
     assert response.status_code == 200
     # must be good (was different IP)
     assert response.content == b'good 2.3.4.5'
@@ -199,7 +192,7 @@ def test_nic_update_authorized_update_other_services(client):
 def test_nic_update_authorized_badagent(client, settings):
     settings.BAD_AGENTS = ['foo', 'bad_agent', 'bar', ]
     response = client.get(reverse('nic_update'),
-                          HTTP_AUTHORIZATION=make_basic_auth_header(TEST_HOST, TEST_SECRET),
+                          HTTP_AUTHORIZATION=make_basic_auth_header(to_http_user(TEST_HOST), TEST_SECRET),
                           HTTP_USER_AGENT='bad_agent')
     assert response.status_code == 200
     assert response.content == b'badagent'
@@ -209,13 +202,13 @@ def test_nic_update_authorized_badip(client, settings):
     settings.BAD_IPS_HOST = IPSet([IPAddress('7.7.7.7'), ])
     # normal update, not on blacklist
     response = client.get(reverse('nic_update') + '?myip=1.2.3.4',
-                          HTTP_AUTHORIZATION=make_basic_auth_header(TEST_HOST, TEST_SECRET))
+                          HTTP_AUTHORIZATION=make_basic_auth_header(to_http_user(TEST_HOST), TEST_SECRET))
     assert response.status_code == 200
     content = response.content.decode('utf-8')
     assert content.startswith('good ') or content.startswith('nochg ')
     # abusive update, ip on blacklist
     response = client.get(reverse('nic_update') + '?myip=7.7.7.7',
-                          HTTP_AUTHORIZATION=make_basic_auth_header(TEST_HOST, TEST_SECRET))
+                          HTTP_AUTHORIZATION=make_basic_auth_header(to_http_user(TEST_HOST), TEST_SECRET))
     assert response.status_code == 200
     assert response.content == b'abuse'
 
@@ -258,31 +251,31 @@ def test_nic_update_session_foreign_host(client):
 
 def test_nic_delete_authorized_invalid_ip1(client):
     response = client.get(reverse('nic_delete') + '?myip=1234',
-                          HTTP_AUTHORIZATION=make_basic_auth_header(TEST_HOST, TEST_SECRET))
+                          HTTP_AUTHORIZATION=make_basic_auth_header(to_http_user(TEST_HOST), TEST_SECRET))
     assert response.status_code == 200
     assert response.content == b'dnserr'
 
 
 def test_nic_delete_authorized_invalid_ip2(client):
     response = client.get(reverse('nic_delete') + '?myip=%C3%A4%C3%BC%C3%B6',
-                          HTTP_AUTHORIZATION=make_basic_auth_header(TEST_HOST, TEST_SECRET))
+                          HTTP_AUTHORIZATION=make_basic_auth_header(to_http_user(TEST_HOST), TEST_SECRET))
     assert response.status_code == 200
     assert response.content == b'dnserr'
 
 
 def test_nic_delete_authorized(client):
     response = client.get(reverse('nic_update') + '?myip=%s' % ('1.2.3.4', ),
-                          HTTP_AUTHORIZATION=make_basic_auth_header(TEST_HOST, TEST_SECRET))
+                          HTTP_AUTHORIZATION=make_basic_auth_header(to_http_user(TEST_HOST), TEST_SECRET))
     assert response.status_code == 200
     response = client.get(reverse('nic_update') + '?myip=%s' % ('::1', ),
-                          HTTP_AUTHORIZATION=make_basic_auth_header(TEST_HOST, TEST_SECRET))
+                          HTTP_AUTHORIZATION=make_basic_auth_header(to_http_user(TEST_HOST), TEST_SECRET))
     assert response.status_code == 200
     response = client.get(reverse('nic_delete') + '?myip=%s' % ('0.0.0.0', ),
-                          HTTP_AUTHORIZATION=make_basic_auth_header(TEST_HOST, TEST_SECRET))
+                          HTTP_AUTHORIZATION=make_basic_auth_header(to_http_user(TEST_HOST), TEST_SECRET))
     assert response.status_code == 200
     assert response.content == b'deleted A'
     response = client.get(reverse('nic_delete') + '?myip=%s' % ('::', ),
-                          HTTP_AUTHORIZATION=make_basic_auth_header(TEST_HOST, TEST_SECRET))
+                          HTTP_AUTHORIZATION=make_basic_auth_header(to_http_user(TEST_HOST), TEST_SECRET))
     assert response.status_code == 200
     assert response.content == b'deleted AAAA'
 

src/nsupdate/api/views.py

@@ -139,18 +139,18 @@ def check_api_auth(username, password, logger=None):
     """
     Check username and password against our database.
 
-    :param username: http basic auth username (== fqdn)
+    :param username: http basic auth username
     :param password: update password
     :return: host object if authenticated, None otherwise.
     """
-    fqdn = username
     try:
-        host = Host.get_by_fqdn(fqdn)
+        host = Host.get_by_http_user(username)
     except ValueError:
         # logging this at debug level because otherwise it fills our logs...
-        logger.debug('%s - received bad credentials (auth username == dyndns hostname not in our hosts DB)' % (fqdn, ))
+        logger.debug('%s - received bad credentials' % (username, ))
         return None
     if host is not None:
+        fqdn = host.get_fqdn()
         ok = check_password(password, host.update_secret)
         success_msg = ('failure', 'success')[ok]
         msg = "api authentication %s. [hostname: %s (given in basic auth)]" % (success_msg, fqdn, )
@@ -218,27 +218,25 @@ def get(self, request, logger=None, delete=False):
             logger.debug('%s - received no auth' % (hostname, ))
             return basic_challenge("authenticate to update DNS", 'badauth')
         username, password = basic_authenticate(auth)
-        if '.' not in username:  # username MUST be the fqdn
-            # specifically point to configuration errors on client side
-            return Response('notfqdn')
-        if username in settings.BAD_HOSTS:
-            return Response('abuse', status=403)
         host = check_api_auth(username, password)
         if host is None:
             return basic_challenge("authenticate to update DNS", 'badauth')
-        logger.info("authenticated by update secret for host %s" % username)
+        fqdn = str(host.get_fqdn())
+        if fqdn in settings.BAD_HOSTS:
+            return Response('abuse', status=403)
+        logger.info("authenticated by update secret for host %s" % fqdn)
         if hostname is None:
             # as we use update_username == hostname, we can fall back to that:
-            hostname = username
-        elif hostname != username:
+            hostname = fqdn
+        elif hostname != fqdn:
             if '.' not in hostname:
                 # specifically point to configuration errors on client side
                 result = 'notfqdn'
             else:
                 # maybe this host is owned by same person, but we can't know.
                 result = 'nohost'  # or 'badauth'?
             msg = ("rejecting to update wrong host %s (given in query string) "
-                   "[instead of %s (given in basic auth)]" % (hostname, username))
+                   "[instead of %s (given in basic auth as user %s)]" % (hostname, fqdn, username))
             logger.warning(msg)
             host.register_client_result(msg, fault=True)
             return Response(result)

src/nsupdate/main/migrations/0013_host_http_user.py

@@ -0,0 +1,38 @@
+# Generated by Django 2.2.9 on 2020-01-10 08:53
+
+from django.db import migrations, models
+import nsupdate.main.models
+
+
+def gen_http_user(apps, schema_editor):
+    Host = apps.get_model('main', 'Host')
+    http_user_set = set()
+    all_hosts = [row for row in Host.objects.all() ]
+    for row in all_hosts:
+        http_user = nsupdate.main.models._http_user_generator()
+        while http_user in http_user_set:
+            http_user = nsupdate.main.models._http_user_generator()
+        http_user_set.add(http_user)
+        row.http_user = http_user
+    for row in all_hosts:
+        row.save(update_fields=['http_user'])
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('main', '0012_auto_20191230_1729'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='host',
+            name='http_user',
+            field=models.CharField(max_length=30, unique=True, verbose_name='http user', null=True, default=None),
+        ),
+        migrations.RunPython(gen_http_user, reverse_code=migrations.RunPython.noop),
+        migrations.AlterField(
+            model_name='host',
+            name='http_user',
+            field=models.CharField(max_length=30, unique=True, verbose_name='http user', null=False, default=nsupdate.main.models._http_user_generator),
+        ),
+    ]