Initial commit.

Author: githubChenchi

CaptivePortal.py

@@ -0,0 +1,197 @@
+import usocket
+import utime
+import _thread
+import dataCall
+import log
+from misc import USBNET
+from misc import Power
+
+class CaptivePortal:
+    def __init__(self, target_url, dns_whitelist, real_dns_server=("8.8.8.8", 53)):
+        """
+        Initializing Captive Portal parameters。
+        :param target_url: The target URL for the redirect (如 "https://example.com/")
+        :param dns_whitelist: A list of DNS whitelists that allow normal resolution
+        :param real_dns_server: Real DNS server address and port
+        """
+        self.target_url = target_url
+        self.dns_whitelist = dns_whitelist
+        self.real_dns_server = real_dns_server
+        self.portal_active = True
+        self.http_thread = None
+        self.dns_thread = None
+        log.basicConfig(level=log.INFO)
+        self.cp_log = log.getLogger("CaptivePortal")
+
+    def get_local_ip(self):
+        """Get the local IP address that will be used to bind the forwarding Socket for DNS requests."""
+        try:
+            return dataCall.getInfo(1, 0)[2][2]
+        except Exception as e:
+            self.cp_log.warning("Error getting local IP: {}".format(e))
+            return "0.0.0.0"
+
+    def forward_to_real_dns(self, request, local_ip):
+        """Forward DNS requests to the real DNS server."""
+        sock = None
+        try:
+            sock = usocket.socket(usocket.AF_INET, usocket.SOCK_DGRAM)
+            sock.settimeout(5)
+            sock.bind((local_ip, 0))
+            sock.sendto(request, self.real_dns_server)
+            response, _ = sock.recvfrom(512)
+            return response
+        except Exception as e:
+            serr = str(e)
+            if serr != "[Errno 113] EHOSTUNREACH":
+                self.cp_log.warning("Error forwarding DNS request: {}".format(e))
+            return None
+        finally:
+            if sock:
+                sock.close()
+
+    def start_http_server(self):
+        """Start the HTTP redirection server."""
+        ip = "192.168.43.1"  # USBNET Gateway IP
+        addr = (ip, 80)
+        s = usocket.socket(usocket.AF_INET, usocket.SOCK_STREAM)
+        s.setsockopt(usocket.SOL_SOCKET, usocket.SO_REUSEADDR, 1)
+        s.bind(addr)
+        s.listen(3)
+        self.cp_log.info("HTTP Server running at http://{}:80/".format(ip))
+
+        while self.portal_active:
+            try:
+                res = s.accept()
+                if not res or len(res) < 2:
+                    continue
+                client_sock = res[0]
+                client_addr = (res[1], res[2])
+                self.cp_log.debug("Connection from {}".format(client_addr))
+
+                try:
+                    request = client_sock.recv(1024).decode('utf-8')
+                    if not request:
+                        self.cp_log.debug("Empty request, closing connection.")
+                        continue
+
+                    self.cp_log.debug("Request: {}".format(request))
+                    response = "HTTP/1.1 302 Found\r\nLocation: {}\r\n\r\n".format(self.target_url)
+                    client_sock.send(response)
+                except Exception as e:
+                    self.cp_log.warning("Request handling error: {}".format(e))
+                finally:
+                    client_sock.close()
+                utime.sleep_ms(100)
+            except Exception as e:
+                self.cp_log.warning("HTTP server error: {}".format(e))
+                continue
+        if client_sock:
+            client_sock.close()
+        s.close()
+
+    def start_dns_server(self):
+        """Start a DNS hijacking server."""
+        ip = "192.168.43.1"  # Captive Portal IP
+        local_ip = self.get_local_ip()
+        dns_sock = usocket.socket(usocket.AF_INET, usocket.SOCK_DGRAM)
+        dns_sock.bind(("0.0.0.0", 53))
+        self.cp_log.info("DNS Server running...")
+
+        while self.portal_active:
+            try:
+                data, addr = dns_sock.recvfrom(1024)
+                self.cp_log.debug("DNS request from {}".format(addr))
+
+                # Resolving DNS requests
+                p = DNSQuery(data)
+
+                if any(whitelisted in p.domain for whitelisted in self.dns_whitelist):
+                    self.cp_log.debug("Domain {} is in whitelist, forwarding to real DNS.".format(p.domain))
+                    response = self.forward_to_real_dns(data, local_ip)
+                    if response is None:
+                        self.cp_log.debug("Failed to get response from real DNS, sending empty response.")
+                        response = b"\x00" * len(data)
+                else:
+                    response = p.response(ip)
+                    self.cp_log.debug('Hijacking: {:s} -> {:s}'.format(p.domain, ip))
+
+                dns_sock.sendto(response, addr)
+            except Exception as e:
+                self.cp_log.warning("DNS handling error: {}".format(e))
+                utime.sleep_ms(100)
+        dns_sock.close()
+
+    def start(self):
+        """Start a Captive Portal, including HTTP and DNS services."""
+        self.portal_active = True
+        self.dns_thread = _thread.start_new_thread(self.start_dns_server, ())
+        self.http_thread = _thread.start_new_thread(self.start_http_server, ())
+        self.cp_log.info("Captive Portal started.")
+
+    def stop(self):
+        """Close the Captive Portal and stop HTTP and DNS services."""
+        self.portal_active = False
+        self.cp_log.info("Captive Portal stopped.")
+
+
+# Resolve the DNS request Class
+class DNSQuery:
+    def __init__(self, data):
+        self.data = data
+        self.domain = ''
+        try:
+            m = data[2]  # Flags
+            tipo = (m >> 3) & 15  # Opcode bits
+            if tipo == 0:
+                ini = 12
+                lon = data[ini]
+                while lon != 0:
+                    self.domain += data[ini+1:ini+lon+1].decode("utf-8") + '.'
+                    ini += lon + 1
+                    lon = data[ini]
+        except Exception as e:
+            self.domain = ''
+            log.warning("Error parsing DNS query: {}".format(e))
+
+    def response(self, ip):
+        """Building a DNS hijacking response"""
+        packet = b''
+        if self.domain:
+            packet += self.data[:2] + b"\x81\x80"
+            packet += self.data[4:6] + self.data[4:6] + b'\x00\x00\x00\x00'
+            packet += self.data[12:]
+            packet += b'\xc0\x0c'
+            packet += b'\x00\x01\x00\x01\x00\x00\x00\x3c\x00\x04'
+            packet += bytes(map(int, ip.split('.')))
+        return packet
+
+# Example
+if __name__ == "__main__":
+    # Windows: Type_RNDIS
+    # Linux/Android/IOS: Type_ECM
+    if (USBNET.get_worktype() != USBNET.Type_RNDIS):
+        USBNET.set_worktype(USBNET.Type_RNDIS)
+        Power.powerRestart()
+    
+    portal = CaptivePortal(
+        target_url="https://python.quectel.com/",
+        dns_whitelist=["www.python.quectel.com", "python.quectel.com"]
+    )
+    portal.start()
+    
+    cnt = 0
+    while True:
+        utime.sleep(1)
+        ret = USBNET.open()
+        cnt = cnt + 1
+        if ret == 0:
+            print("USBNET status: ",USBNET.get_status())
+            print("USBNET type: ",USBNET.get_worktype())
+            break
+        if cnt == 60:
+            print("USBNET open fail!")
+            portal.stop()
+            break
+    
+

LICENSE

@@ -0,0 +1,13 @@
+Copyright (c) Quectel Wireless Solution, Co., Ltd.All Rights Reserved.
+ 
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+ 
+    http://www.apache.org/licenses/LICENSE-2.0
+ 
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.

readme.md

@@ -0,0 +1,43 @@
+# CaptivePortal lib for QuecPython
+
+## Overview
+
+**Captive Portal** is a network access control mechanism commonly used in public Wi-Fi networks. When users connect to a network requiring authentication, they are automatically redirected to a webpage where they must enter a username and password or accept the terms of use. In addition to authentication, Captive Portals can also be used for displaying advertisements, informing users of network usage policies, and more.
+
+Integrating **Captive Portal functionality into a USB dongle** allows you to manage devices connected to your network more conveniently and provide a personalized network experience. This feature can enhance security, enable user interaction through customized portals, and facilitate the deployment of network usage policies, advertisements, or service announcements.
+
+**QuecPython** provides an implementation of the Captive Portal technology, allowing users to specify the webpage they want the Captive Portal to redirect to. This enables users to implement their own network management strategies effectively.
+
+## Example
+
+For demonstration purposes, this example specifies **https://python.quectel.com** as the target page for Captive Portal redirection.
+
+```python
+# Example
+if __name__ == "__main__":
+    # Windows: Type_RNDIS
+    # Linux/Android/IOS: Type_ECM
+    if (USBNET.get_worktype() != USBNET.Type_RNDIS):
+        USBNET.set_worktype(USBNET.Type_RNDIS)
+        Power.powerRestart()
+    
+    portal = CaptivePortal(
+        target_url="https://python.quectel.com/",
+        dns_whitelist=["www.python.quectel.com", "python.quectel.com"]
+    )
+    portal.start()
+    
+    cnt = 0
+    while True:
+        utime.sleep(1)
+        ret = USBNET.open()
+        cnt = cnt + 1
+        if ret == 0:
+            print("USBNET status: ",USBNET.get_status())
+            print("USBNET type: ",USBNET.get_worktype())
+            break
+        if cnt == 60:
+            print("USBNET open fail!")
+            portal.stop()
+            break
+```