(request, settings_key, user=None)
| 33 | |
| 34 | |
| 35 | def network_access_allowed(request, settings_key, user=None): |
| 36 | try: |
| 37 | network_access = CoreSettings.objects.get(key=NETWORK_ACCESS_KEY).value |
| 38 | except CoreSettings.DoesNotExist: |
| 39 | network_access = {} |
| 40 | local_cidrs = ["127.0.0.0/8", "10.0.0.0/8", "172.16.0.0/12", "192.168.0.0/16", "::1/128", "fc00::/7", "fe80::/10"] |
| 41 | # Set defaults based on endpoint type |
| 42 | if settings_key == "M3U_EPG": |
| 43 | # M3U/EPG endpoints: local IPv4 and IPv6 only by default |
| 44 | default_cidrs = local_cidrs |
| 45 | else: |
| 46 | # Other endpoints: allow all by default |
| 47 | default_cidrs = ["0.0.0.0/0", "::/0"] |
| 48 | |
| 49 | cidrs = ( |
| 50 | network_access[settings_key].split(",") |
| 51 | if settings_key in network_access |
| 52 | else default_cidrs |
| 53 | ) |
| 54 | |
| 55 | network_allowed = False |
| 56 | client_ip = ipaddress.ip_address(get_client_ip(request)) |
| 57 | for cidr in cidrs: |
| 58 | network = ipaddress.ip_network(cidr) |
| 59 | if client_ip in network: |
| 60 | network_allowed = True |
| 61 | break |
| 62 | |
| 63 | if not network_allowed: |
| 64 | return False |
| 65 | |
| 66 | if user is not None: |
| 67 | user_networks = (getattr(user, 'custom_properties', None) or {}).get('allowed_networks', {}) |
| 68 | raw = user_networks.get(settings_key, '') |
| 69 | if raw: |
| 70 | for cidr in (c.strip() for c in raw.split(',') if c.strip()): |
| 71 | try: |
| 72 | if client_ip in ipaddress.ip_network(cidr, strict=False): |
| 73 | return True |
| 74 | except ValueError: |
| 75 | continue |
| 76 | return False |
| 77 | |
| 78 | return True |
no test coverage detected