gandalf/tests/test_app.py

"""Tests for app.py routes: database and LDAP are mocked."""
import json
import os
import sys
from unittest.mock import patch

import pytest

sys.path.insert(0, os.path.dirname(os.path.dirname(__file__)))

import app as gandalf  # noqa: E402

PROXY = '10.0.0.27'
# What the local nginx forwards for a request that came through the Authelia proxy
AUTH = {'Remote-User': 'jared', 'Remote-Name': 'Jared V', 'Remote-Groups': 'admin', 'X-Real-IP': PROXY}
CONFIG = {'auth': {'allowed_groups': ['admin'], 'trusted_proxies': [PROXY]}, 'ticket_api': {}}
STATE = {
    'network_snapshot': json.dumps({
        'hosts': {'large1': {'ip': '10.0.0.2', 'status': 'up', 'source': 'prometheus', 'interfaces': {'eth0': 'up'}}},
        'unifi': [{'name': 'USW Flex', 'type': 'usw', 'model': 'USF5P', 'ip': '10.0.0.3', 'mac': '',
                   'connected': True}],
    }),
    'link_stats': json.dumps({'hosts': {}, 'unifi_switches': {}}),
}


@pytest.fixture
def client():
    with patch.object(gandalf, 'db') as db:
        db.get_state.side_effect = lambda key, default=None: STATE.get(key, default)
        db.get_active_events.return_value = []
        db.get_recent_resolved.return_value = []
        db.get_active_suppressions.return_value = []
        db.get_events_for_target.return_value = []
        db.count_active_events.return_value = 0
        db.get_status_summary.return_value = {'critical': 0, 'warning': 0, 'info': 0, 'resolved_24h': 0}
        db.check_suppressed.return_value = False
        with patch.object(gandalf, '_cfg', CONFIG):
            yield gandalf.app.test_client()


class TestIdentityHeaderTrust:
    def test_proxy_identity_is_accepted(self, client):
        assert client.get('/api/devices', headers=AUTH).status_code == 200

    def test_lan_client_through_nginx_cannot_claim_a_user(self, client):
        spoofed = dict(AUTH, **{'X-Real-IP': '10.0.0.99'})
        assert client.get('/api/devices', headers=spoofed).status_code == 401

    def test_loopback_without_nginx_source_is_untrusted(self, client):
        spoofed = {k: v for k, v in AUTH.items() if k != 'X-Real-IP'}
        assert client.get('/api/devices', headers=spoofed).status_code == 401

    def test_direct_peer_cannot_borrow_the_proxy_address(self, client):
        resp = client.get('/api/devices', headers=AUTH, environ_base={'REMOTE_ADDR': '10.0.0.99'})
        assert resp.status_code == 401

    def test_proxy_connecting_directly_is_trusted(self, client):
        resp = client.get('/api/devices', headers=AUTH, environ_base={'REMOTE_ADDR': PROXY})
        assert resp.status_code == 200

    def test_fails_closed_without_trusted_proxies(self, client):
        with patch.object(gandalf, '_cfg', {'auth': {'allowed_groups': ['admin']}}):
            assert client.get('/api/devices', headers=AUTH).status_code == 401

    def test_cidr_entries_and_bad_entries(self, client):
        cfg = {'auth': {'allowed_groups': ['admin'], 'trusted_proxies': ['not-an-ip', '10.0.0.0/24']}}
        with patch.object(gandalf, '_cfg', cfg):
            assert client.get('/api/devices', headers=AUTH).status_code == 200

    def test_garbage_source_header_is_untrusted(self, client):
        assert client.get('/api/devices', headers=dict(AUTH, **{'X-Real-IP': 'not-an-ip'})).status_code == 401

    def test_rejection_is_logged(self, client, caplog):
        client.get('/api/devices', headers=dict(AUTH, **{'X-Real-IP': '10.0.0.99'}))
        assert "untrusted source '10.0.0.99'" in caplog.text

    def test_avatar_ignores_untrusted_identity(self, client):
        with patch.object(gandalf, '_config', return_value=dict(CONFIG, ldap={'host': 'h', 'bind_dn': 'b'})):
            resp = client.get('/api/avatar', headers=dict(AUTH, **{'X-Real-IP': '10.0.0.99'}))
        assert resp.status_code == 401


class TestPagesAndNav:
    def test_home_has_search_and_no_link_debug(self, client):
        html = client.get('/', headers=AUTH).get_data(as_text=True)
        assert 'id="device-search-input"' in html
        assert 'Link Debug' not in html

    def test_home_section_order(self, client):
        html = client.get('/', headers=AUTH).get_data(as_text=True)
        order = [html.index(m) for m in (
            'id="device-search"', 'Alert Queue', 'WAN / Fiber', 'g-section-title">Hosts<',
            'Device Inventory', 'Network Topology',
        )]
        assert order == sorted(order)

    def test_old_links_url_redirects_home(self, client):
        resp = client.get('/links', headers=AUTH)
        assert resp.status_code == 302 and resp.headers['Location'].endswith('/')

    def test_device_page_escapes_name(self, client):
        html = client.get('/device/%3Cscript%3Ex', headers=AUTH).get_data(as_text=True)
        assert '<script>x' not in html

    def test_old_inspector_url_redirects_home(self, client):
        resp = client.get('/inspector', headers=AUTH)
        assert resp.status_code == 302 and resp.headers['Location'].endswith('/')

    def test_nav_has_no_inspector_link(self, client):
        assert '>Inspector<' not in client.get('/', headers=AUTH).get_data(as_text=True)
        assert 'Inspector' not in client.get('/device/large1', headers=AUTH).get_data(as_text=True).split('</nav>')[0]

    def test_device_page_loads_the_port_inspector(self, client):
        html = client.get('/device/large1', headers=AUTH).get_data(as_text=True)
        assert 'port-inspector.js' in html and 'id="inspector-panel"' in html

    def test_faceplates_load_before_the_port_inspector(self, client):
        html = client.get('/device/large1', headers=AUTH).get_data(as_text=True)
        assert html.index('faceplates.js') < html.index('port-inspector.js')

    def test_link_data_api_remains(self, client):
        assert client.get('/api/links', headers=AUTH).status_code == 200


class TestDeviceApi:
    def test_requires_auth(self, client):
        assert client.get('/api/devices').status_code == 401
        assert client.get('/api/device/large1').status_code == 401

    def test_index_lists_devices(self, client):
        names = [d['name'] for d in client.get('/api/devices', headers=AUTH).get_json()['devices']]
        assert names == ['large1', 'USW Flex']

    def test_name_with_a_space(self, client):
        d = client.get('/api/device/USW%20Flex', headers=AUTH).get_json()
        assert d['kind'] == 'switch' and d['events'] == []

    def test_unknown_device_is_404(self, client):
        assert client.get('/api/device/nope', headers=AUTH).status_code == 404


class TestAvatarLogging:
    def test_missing_ldap_config_says_why(self, client, caplog):
        assert client.get('/api/avatar', headers=AUTH).status_code == 404
        assert 'no ldap' in caplog.text


class TestDatabaseTimeZone:
    def test_sessions_run_in_utc(self):
        import db as real_db
        with patch.object(real_db.pymysql, 'connect') as connect:
            real_db._new_conn({'host': 'h', 'user': 'u', 'password': 'p', 'name': 'n'})
        assert connect.call_args.kwargs['init_command'] == "SET time_zone='+00:00'"


class TestDiagnoseWithResolvedNeighbor:
    LINKS = json.dumps({
        'hosts': {'large1': {'enp35s0': {'host_ip': '10.0.0.2'}}},
        'unifi_switches': {'sw': {'ports': {'Port 4': {
            'port_idx': 4, 'up': True,
            'lldp': {'system_name': 'large1', 'port_id': 'enp35s0', 'chassis_id': 'aa:bb:cc:dd:ee:ff'},
        }}}},
    })

    def test_port_with_a_named_neighbor_starts_a_job(self, client):
        STATE['link_stats'], saved = self.LINKS, STATE['link_stats']
        try:
            with patch.object(gandalf.threading, 'Thread'):
                resp = client.post('/api/diagnose', json={'switch_name': 'sw', 'port_idx': 4}, headers=AUTH)
        finally:
            STATE['link_stats'] = saved
        assert resp.status_code == 200 and resp.get_json()['job_id']