parker/tl841: add tests for info-pull

parker/conftest.py

 import pytest
 import yaml
 import re
+import os
 
 def pytest_configure(config):
     config.addinivalue_line("markers", "config_mode")
@@ -46,6 +47,37 @@ def host_container(target):
     target.deactivate(shell)
     target.deactivate(container)
 
+@pytest.fixture(scope="module")
+def info_pull_ssh_key(target):
+    """
+    This fixture sets up our known ssh-key for the info-pull user.
+    """
+
+    command = target.get_driver("ShellDriver", name="dut_shell")
+    auth_keys_file = "/usr/share/info-pull/.ssh/authorized_keys"
+    # read previous file
+    oldkey, _, rc = command.run(f"cat {auth_keys_file}")
+    assert rc == 0, "Could not read previous info-pull authorized_keys file"
+
+    # read our ssh pubkey
+    keyfile = ssh_dir = os.path.join(
+        os.environ['HOME'],
+        ".ssh",
+        "id_rsa.pub",
+        )
+    with open(keyfile) as fh:
+        pubkey = fh.read()
+
+    # append our key with the original command
+    assert len(oldkey) >= 1, "Could not read previous command part"
+    assert " ssh-" in oldkey[0], "Could not read previous command part"
+    command_part = oldkey[0].split(" ssh-")[0]
+    newkey = f"{command_part} {pubkey}"
+
+    _, _, rc = command.run(f"echo '{newkey}' >> {auth_keys_file}")
+    assert rc == 0
+
+
 def ipv6_link_local(command):
     stdout, _, code = command.run('ip address show br-client')
     if code != 0:

parker/test_parker.py

@@ -50,7 +50,6 @@ def test_user_has_id_rsa():
     assert os.path.isfile(os.path.join(ssh_dir, "id_rsa.pub")), "User has no ~/.ssh/id_rsa.pub"
     
 def test_uboot(target, in_uboot):
-    #command = target.get_driver(CommandProtocol)
     command = target.get_driver('UBootDriver')
 
     stdout, stderr, returncode = command.run('version')
@@ -60,7 +59,6 @@ def test_uboot(target, in_uboot):
     assert 'U-Boot' in '\n'.join(stdout)
 
 def test_help(target, in_uboot):
-    #command = target.get_driver(CommandProtocol)
     command = target.get_driver('UBootDriver')
 
     stdout, stderr, returncode = command.run('help')
@@ -117,6 +115,21 @@ def test_ssh_authorized_keys(target, in_good_config, command):
     else:
         return pytest.skip('Version is neither beta nor stable')
 
+@pytest.mark.config_mode
+def test_info_pull_auth_keys_file(command):
+    auth_keys_file = "/usr/share/info-pull/.ssh/authorized_keys"
+    _, _, rc = command.run(f"ls {auth_keys_file}")
+    assert rc == 0, "info_pull authorized_keys file not found"
+
+@pytest.mark.config_mode
+def test_info_pull_auth_keys_file(command):
+    auth_keys_file = "/usr/share/info-pull/.ssh/authorized_keys"
+    keys_file, _, rc = command.run(f"cat {auth_keys_file}")
+    assert rc == 0, "info_pull authorized_keys file not found"
+
+    for line in keys_file:
+        assert line.startswith("command=")
+
 @pytest.mark.running_mode
 def test_transition_to_running_mode(in_good_running):
     pass
@@ -194,6 +207,88 @@ def test_generated_wg_key(target, in_good_running, command):
     stdout, _, code = command.run('cat /etc/parker/wg-privkey | wg pubkey')
     assert code == 0, "Verification of WG private key failed: {}".format(''.join(stdout))
 
+@pytest.mark.running_mode
+@pytest.mark.flaky(reruns=10, reruns_delay=10)
+def test_good_running_info_pull_noKey(target, in_good_running, command):
+    ip = ipv6_link_local(command)
+    ifname = target.get_resource("RemoteNetworkInterface").ifname
+    keyfile= os.path.join(
+            os.environ['HOME'],
+            ".ssh",
+            "id_rsa",
+            )
+    r = subprocess.run([
+        "ssh",
+        "-o", "UserKnownHostsFile=/dev/null",
+        "-o", "StrictHostKeyChecking=no",
+        "-o", "PreferredAuthentications=publickey",
+        "-i", keyfile,
+        'info-pull@{}%{}'.format(ip, ifname),
+        ])
+    assert r.returncode == 255, "SSH to info_pull without key worked!"
+
+@pytest.mark.running_mode
+@pytest.mark.flaky(reruns=10, reruns_delay=10)
+def test_good_running_info_pull_withKey(target, in_good_running, command, info_pull_ssh_key):
+    # generate a info file even if the cronjob has not run yet
+    _, _, rc = command.run("/usr/share/info-pull/info-export.sh")
+
+    ip = ipv6_link_local(command)
+    ifname = target.get_resource("RemoteNetworkInterface").ifname
+    keyfile= os.path.join(
+            os.environ['HOME'],
+            ".ssh",
+            "id_rsa",
+            )
+    r = subprocess.run([
+        "ssh",
+        "-o", "UserKnownHostsFile=/dev/null",
+        "-o", "StrictHostKeyChecking=no",
+        "-o", "PreferredAuthentications=publickey",
+        "-i", keyfile,
+        'info-pull@{}%{}'.format(ip, ifname),
+        ],
+        check=True,
+        stdout=subprocess.PIPE,
+    )
+
+    assert r.returncode == 0
+    assert "ffci-1" in r.stdout.decode()
+
+@pytest.mark.running_mode
+@pytest.mark.flaky(reruns=10, reruns_delay=10)
+def test_good_running_info_pull_command_fencing(target, in_good_running, command, info_pull_ssh_key):
+    ip = ipv6_link_local(command)
+    ifname = target.get_resource("RemoteNetworkInterface").ifname
+    keyfile= os.path.join(
+            os.environ['HOME'],
+            ".ssh",
+            "id_rsa",
+            )
+    r = subprocess.run([
+        "ssh",
+        "-o", "UserKnownHostsFile=/dev/null",
+        "-o", "StrictHostKeyChecking=no",
+        "-o", "PreferredAuthentications=publickey",
+        "-i", keyfile,
+        'info-pull@{}%{}'.format(ip, ifname),
+        "echo", "freifunkIstCool",
+        ],
+        check=True,
+        stdout=subprocess.PIPE,
+    )
+
+    assert r.returncode == 0
+    assert "freifunkIstCool" not in r.stdout.decode()
+
+@pytest.mark.running_mode
+def test_good_running_info_pull_nonLegacy(target, in_good_running, command):
+
+    stdout, _, rc = command.run("uci show gluon-node-info")
+    assert rc == 0
+
+    for line in stdout:
+        assert ".contact=" not in line
 
 @pytest.mark.running_mode
 @pytest.mark.flaky(reruns=20, reruns_delay=10)

parker/tl841.py

@@ -172,7 +172,8 @@ class SmallUBootStrategy(Strategy):
                 "id.1":"1",
                 "id.1.4.hostname": "ci_841_v11",
                 "id.1.5.meshvpn":"1",
-                "id.1.6.contact": ""
+                "id.1.6.contact": "Freifunk Braunschweig CI ffci-1",
+                "id.1.7.contact": "Freifunk Braunschweig CI ffci-1",
             }
             r = requests.post("http://192.168.1.1/cgi-bin/config/wizard", params=payload)