深澜自动登录Python脚本分享

引言

我们学校用的是深澜认证来验证登录校园网的，因为有无图形化的设备自动进行校园网认证的需求，因此通过分析js以及web认证过程的方式用Python写了一个只依赖最基础Python库的登陆脚本，这个脚本在校内2年稳定运行，因为看到网上没有类似的东西，因此分享给大家

脚本内容

#coding:utf-8
import requests
import json
import base64
import math
import sys,os
import time


class NekoBase64:
    def __init__(self,Alpha=None):
        if Alpha is None or len(Alpha) != 64:
            self.array = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/"
        else:
            self.array = Alpha

    def b64encode(self, str):
        from io import StringIO
        # fill with zero
        length = len( str )
        remainder = length % 3

        if( remainder == 1 ):
            str = str + b'\x00\x00'  # add twice
            length += 2
        elif( remainder == 2 ):
            str = str + b'\x00'  # add once
            length += 1

        # encode
        i = 0
        buf = StringIO()
        if( remainder > 0 ):        
            # if zero is filled in tail
            while i < length - 3:
                en = self._b64encode_str( str[ i ], str[ i+1 ], str[ i+2 ] )
                buf.write( en )
                i += 3
            # print( remainder, i, buf.getvalue() )
            en = self._b64encode_str( str[ i ], str[ i+1 ], str[ i+2 ] )
            buf.write( en[ 0 ] )
            buf.write( en[ 1 ] )

            if( remainder == 2 ):
                buf.write( en[ 2 ] )  # add once
                buf.write( '=' )
            elif( remainder == 1 ):
                buf.write( '==' )   # add twice
        else:  # remainder == 0
            while i < length:
                en = self._b64encode_str( str[ i ], str[ i+1 ], str[ i+2 ] )
                buf.write( en )
                i += 3
        
        # print( strlist )
        return buf.getvalue().encode()

    def _b64encode_str(self, s0, s1, s2 ):
        d = s2 & 63
        d = self.array[ d ]
        c1 = ( s1 & 15  ) << 2
        c2 = ( s2 & 192 ) >> 6
        c = c1 + c2
        c = self.array[ c ]
        b1 = ( s0   & 3 ) << 4
        b2 = ( s1 & 240 ) >> 4
        b = b1 + b2
        b = self.array[ b ]
        a = ( s0 & 252 ) >> 2
        a = self.array[ a ]
        return ''.join( [ a, b, c, d ] )

class ShenLanLogin:
    enc = "s" + "run" + "_bx1"
    n = '200'
    types = '1'
    _ALPHA = 'LVoJPiCN2R8G90yg+hmFHuacZ1OWMnrsSTXkYpUq/3dlbfKwv6xztjI7DeBE45QA'
    real_name = ""
    srun_ver = ""
    wallet_balance = 0
    __login_status = False
    server_ip = ""
    error_info = {
        "E0000": "登录成功",
        "E2401": "User-Request",
        "E2402": "Lost-Carrier",
        "E2404": "Idle-Timeout",
        "E2405": "Session-Timeout",
        "E2406": "Admin-Reset",
        "E2407": "Admin-Reboot",
        "E2408": "Port-Error",
        "E2409": "NAS-Error",
        "E2410": "NAS-Request",
        "E2411": "NAS-Reboot",
        "E2412": "Port-Unneeded",
        "E2413": "Port-Preempted",
        "E2414": "Port-Suspended",
        "E2415": "Service-Unavailable",
        "E2416": "Callback",
        "E2417": "User-Error",
        "E2531": "用户不存在",
        "E2532": "您的两次认证的间隔太短,请稍候10秒后再重试登录",
        "E2533": "密码错误次数超过限制，请5分钟后再重试登录",
        "E2534": "有代理行为被暂时禁用",
        "E2535": "认证系统已经被禁用",
        "E2536": "授权已过期",
        "E2553": "帐号或密码错误",
        "E2601": "该区域仅允许手机登录。",
        "E2602": "您还没有绑定手机号或绑定的非联通手机号码",
        "E2606": "用户被禁用",
        "E2607": "接口被禁用",
        "E2611": "您当前使用的设备非该账号绑定设备 请绑定或使用绑定的设备登入",
        "E2613": "NAS PORT绑定错误",
        "E2614": "MAC地址绑定错误",
        "E2615": "IP地址绑定错误",
        "E2616": "用户已欠费",
        "E2620": "已经在线了",
        "E2621": "已经达到授权人数",
        "E2806": "找不到符合条件的产品",
        "E2807": "找不到符合条件的计费策略",
        "E2808": "找不到符合条件的控制策略",
        "E2833": "IP不在DHCP表中，需要重新拿地址。",
        "E2840": "校内地址不允许访问外网",
        "E2841": "IP地址绑定错误",
        "E2842": "IP地址无需认证可直接上网",
        "E2843": "IP地址不在IP表中",
        "E2844": "IP地址在黑名单中",
        "E2901": "密码错误",
        "E6500": "认证程序未启动",
        "E6501": "用户名输入错误",
        "E6502": "注销时发生错误，或没有帐号在线",
        "E6503": "您的账号不在线上",
        "E6504": "注销成功，请等1分钟后登录",
        "E6505": "您的MAC地址不正确",
        "E6506": "用户名或密码错误，请重新输入",
        "E6507": "您无须认证，可直接上网",
        "E6508": "您已欠费，请尽快充值",
        "E6509": "您的资料已被修改正在等待同步，请2钟分后再试。如果您的帐号允许多个用户上线，请到WEB登录页面注销",
        "E6510": "您的帐号已经被删除",
        "E6511": "IP已存在，请稍后再试",
        "E6512": "在线用户已满，请稍后再试",
        "E6513": "正在注销在线账号，请重新连接",
        "E6514": "你的IP地址和认证地址不附，可能是经过小路由器登录的",
        "E6515": "系统已禁止客户端登录，请使用WEB方式登录",
        "E6516": "您的流量已用尽",
        "E6517": "您的时长已用尽",
        "E6518": "您的IP地址不合法，可能是：一、与绑的IP地址附；二、IP不允许在当前区域登录",
        "E6519": "当前时段不允许连接",
        "E6520": "抱歉，您的帐号已禁用",
        "E6521": "您的IPv6地址不正确，请重新配置IPv6地址",
        "E6522": "客户端时间不正确，请先同步时间（或者是调用方传送的时间格式不正确，不是时间戳；客户端和服务器之间时差超过2小时，括号里面内容不要提示给客户）",
        "E6523": "认证服务无响应",
        "E6524": "计费系统尚未授权，目前还不能使用",
        "E6525": "后台服务器无响应;请联系管理员检查后台服务运行状态",
        "E6526": "您的IP已经在线;可以直接上网;或者先注销再重新认证",
        "E6527": "当前设备不在线",
        "E6528": "您已经被服务器强制下线",
        "E6529": "身份验证失败，但不返回错误消息",
        0:"本机IP已经使用其他账号登陆在线了"
    }

    def __init__(self,username="",password=None,password_base64=None, acid="4",server_ip="192.168.118.51",user_ip="218.195.213.250"):
        self.username = username
        self.ip = user_ip
        self.acid = acid
        self.base64 =  NekoBase64(self._ALPHA)
        self.server_ip = server_ip
        if password is None:
            import base64
            self.password = base64.b64decode(password_base64).decode()
        else:
            self.password = password
    def Logger(self,status,info,content=""):
        datetime_string = time.strftime('%Y-%m-%d %H:%M:%S',time.localtime(time.time()))
        print(status,datetime_string,info,(str(content) if len(str(content)) < 30 else str(content)[:30]))

    def login(self):
        addr = "http://%s/cgi-bin/get_challenge?callback=json&username=%s&ip=%s"
        addr = addr%(self.server_ip,self.username,self.ip)
        data = requests.get(addr).content[5:-1].decode()
        data = json.loads(data)
        self.srun_ver = data['srun_ver']
        self.Logger("[*]","认证系统版本:",self.srun_ver)
        token = data['challenge']   #token = challenge
        self.Logger("[*]","获取认证 Token",token)
        i = self.info({
        			'username': self.username,
        			'password': self.password,
        			'ip': self.ip,
        			'acid': self.acid,
        			'enc_ver': self.enc
        	}, token)
        self.Logger("[*]","生成认证 INFO",i.decode())
        hmd5 = self.pwd(self.password, token)
        self.Logger("[*]","生成校验 HMD5",hmd5)
        chkstr = token.encode("utf-8") + self.username.encode("utf-8")
        chkstr += token.encode("utf-8") + hmd5.encode("utf-8")
        chkstr += token.encode("utf-8") + self.acid.encode("utf-8")
        chkstr += token.encode("utf-8") + self.ip.encode("utf-8")
        chkstr += token.encode("utf-8") + self.n.encode("utf-8")
        chkstr += token.encode("utf-8") + self.types.encode("utf-8")
        chkstr += token.encode("utf-8") + i
        OS = {
                'device':'Windows 10',
                'platform':'Windows'
            }
        params = {
                # 'callback':'jQuery112405101259630772381_1603749911530',
                'callback':"json",
        		'action': "login",
        		'username': self.username,
        		'password': "{MD5}"+hmd5,
        		'ac_id': self.acid,
        		'ip': self.ip,
        		'chksum': self.chksum(chkstr),
        		'info': i,
        		'n': self.n,
        		'type': self.types,
        		'os': OS['device'],
        		'name': OS['platform'],
                'double_stack':"1"
        	}
        status,json_data = self.srunPortal(params)
        self.__login_status = status
        if status:
            self.Logger("[+]","网络登陆通过")
            self.real_name = json_data['real_name']
            self.wallet_balance = json_data['wallet_balance']
            self.srun_ver = json_data['srun_ver']
        else:
            try:
                if json_data['error'] == "ip_already_online_error" or json_data['suc_msg'] == "ip_already_online_error":
                    self.Logger("[*]","登陆有点小问题，自动处理问题:",self.error_info[json_data['ecode']])
                    self.Logger("[+]","自动下线并重新登陆")
                    self.Logout(json_data["online_ip"])
                    status,json_data = self.srunPortal(params)
                    self.__login_status = status
                    if status:
                        self.Logger("[+]","网络登陆通过")
                        self.real_name = json_data['real_name']
                        self.wallet_balance = json_data['wallet_balance']
                        self.srun_ver = json_data['srun_ver']
                    else:
                        self.Logger("[-]","登陆失败,失败原因",self.error_info[json_data['ecode']])
                else:
                    self.Logger("[-]","登陆失败,失败原因",self.error_info[json_data['ecode']])
            except:
                self.Logger("[-]","登陆失败,失败原因",json_data["res"])
                print(json_data)
    
    def srunPortal(self,params):
        headers = {"User-Agent":"Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:81.0) Gecko/20100101 Firefox/81.0"}
        addr = "http://192.168.118.51/cgi-bin/srun_portal"
        r = requests.get(addr,params=params,headers=headers)
        response = r.content[5:-1].decode()
        import json
        ret_json = json.loads(response)
        if ret_json['error'] == "ok" and ret_json['suc_msg'] == "login_ok":
            return True,ret_json
        else:
            return False,ret_json

    def pwd(self,a,b):
        a = a.encode("utf-8")
        b = b.encode("utf-8")
        import hmac
        h = hmac.new(b, a, digestmod='MD5')
        return h.hexdigest()

    def chksum(self,chkstr):
        import hashlib
        sha1 = hashlib.sha1()
        sha1.update(chkstr)
        return sha1.hexdigest()

    def s(self, a, b):
        c = len(a)
        a = a + "\x00"*(4-c%4)
        c = len(a)
        v = []
        i = 0
        while i<c:
            v.append(ord(a[i]) | ord(a[i+1])<<8 | ord(a[i+2]) << 16 | ord(a[i+3]) << 24)
            i += 4
        if b:
            v.append(c)
        return v

    def json(self,d):
        return json.dumps(d)

    def info(self, d, k):
        return b"{SRBX1}" + self.base64.b64encode(self.xEncode(self.json(d), k))

    def l(self,a, b):
        d = len(a) 
        c = 0xffffffff & ((d - 1) << 2)
        if b:
            m = a[d - 1]
            if ((m < c - 3) or (m > c)):
                return None
            c = m
        # for (var i = 0; i < d; i++) {
        for i in range(d):
            a[i] = b"%c%c%c%c"%(a[i] & 0xff, a[i] >> 8 & 0xff, a[i] >> 16 & 0xff, a[i] >> 24 & 0xff)
        if b:
            return (b''.join(a))[0:c]
        else:
            return b''.join(a)
        """
        uSXafDDmjIPKqHtMC8LqJNmnoRssnMzy1MWLRuVxXyoOMJ/37VJf0m42Bst5vd01VlVz5t2BxpJyu0d88DwNWlEHsUkm5uGyRbRAPmlF8Pe6Wc4YUe4i2OBxJkrK1jreMS6GOKWPv3XXFXEFZgmGjS==
        TJAA0dOVmws/luI+XlqsAkL/iIpN/ZQhDdZq07QV5VafDkpS2YLYl4RmsZE7yb3B0AVSeCbrrby6g+VbpbDuhPUOUFfkVCwiNZ/HIHjO5uUDvSANHsQ2NDqcVWfSlyQtDyaJqryPEicHH96Hrb6c6S==
        """

    def xEncode(self,strs, key): 
        from bitstring import BitArray
        if strs == "":
            return ""
        v = self.s(strs, True)
        k = self.s(key, False)
        if len(k) < 4:
            for _ in range(4-len(k)):
                k.append(0)
        # if (len < 4) {
        #     k.length = 4;
        # }
        n = len(v) - 1


        #z = v[n]
        print(v[n])
        z = BitArray("uint:32=%d"%v[n],length=32)
        
        y = BitArray(32)
        y.int = v[0]
        c = BitArray("0x86014019",length=32) | BitArray("0x183639A0",length=32)
        m = BitArray("0",length=32)
        e = BitArray("0",length=32)
        p = BitArray("0",length=32)
        q = math.floor(6 + 52 / (n + 1))
        d = BitArray("uint:32=0",length=32)
        #####
        #while (0 < q--) 
        ####
        q -= 1
        while 0 <= q:

            d.uint = (d.uint + (c & (BitArray("0x8CE0D9BF",length=32) | BitArray("0x731F2640",length=32))).uint)&0xffffffff
            e = d >> 2 & BitArray("uint:32=3",length=32)

            p = 0
            while p < n:
                y.int = v[p + 1]

                # m = z >>> 5 ^ y << 2;
                m = z >> 5 ^ y << 2

                # m += (y >>> 3 ^ z << 4) ^ (d ^ y);
                m.uint = (((y >> 3 ^ z << 4) ^ (d ^ y)).uint + m.uint)&0xffffffff

                # m += k[(p & 3) ^ e] ^ z;
                tmp_p = BitArray("uint:32=%d"%p,length=32)
                index = (tmp_p & BitArray("uint:32=3",length=32)) ^ e
                if index.uint < len(k):
                    m.uint = ((BitArray("uint:32=%d"%k[index.uint],length=32) ^ z).uint+m.uint)&0xffffffff
                else:
                    m.uint = (z.uint+m.uint)&0xffffffff



                # z = v[p] = v[p] + m & (0xEFB8D130 | 0x10472ECF);
                z  = BitArray("uint:32=%d"%( (v[p] + m.uint )&0xffffffff ),length=32) &\
                     (BitArray("0xEFB8D130",length=32) | BitArray("0x10472ECF",length=32))
                v[p] = z.int

                p += 1
            y = BitArray("uint:32=%d"%(v[0]&0xffffffff),length=32)

            #m = z >>> 5 ^ y << 2
            m = z >> 5 ^ y << 2

            #m += (y >>> 3 ^ z << 4) ^ (d ^ y)
            m.uint = (((y >> 3 ^ z << 4) ^ (d ^ y)).uint + m.uint)&0xffffffff
            
            # m = -1112481212
            # p = 22  e = 2 z = -515815031
            # k = [1869309294, 1768776043]
            # m += k[(p & 3) ^ e] ^ z
            tmp_p = BitArray("uint:32=%d"%p,length=32)
            index = (tmp_p & BitArray("uint:32=3",length=32)) ^ e
            if index.uint < len(k):
                k_tmp = BitArray(32)
                k_tmp.int = k[index.uint]
                m.uint = 0xffffffff & ((k_tmp^z).uint + m.uint)
            else:
                m.uint = (z.uint+m.uint)&0xffffffff

            #z = v[n] = v[n] + m & (0xBB390742 | 0x44C6F8BD);
            z  = BitArray("uint:32=%d"%( (v[n] + m.uint )&0xffffffff ),length=32) &\
                     (BitArray("0xBB390742",length=32) | BitArray("0x44C6F8BD",length=32))
            v[n] = z.int
            q -= 1
        return self.l(v, False)

    def Logout(self,online_ip):
        addr = "http://%s/cgi-bin/srun_portal?callback=json&action=logout&ac_id=%s&ip=%s&username=%s"%(self.server_ip,self.acid,online_ip,self.username)
        jsdata = json.loads(requests.get(addr).content[5:-1])
        self.__login_status = False
        if jsdata["error"] == "ok":
            self.Logger("[+]","自动下线成功")
            return True
        else:
            self.Logger("[-]","自动下线失败，失败原因:",jsdata['error'])
            return False

    def Show_User_Info(self):
        if not self.__login_status:
            self.Logger("[-]","网络认证未登陆或认证失败")
            return
        self.Logger("[*]","用户名:",self.username)
        self.Logger("[*]","姓名:",self.real_name)
        self.Logger("[*]","IP:",self.ip)
        self.Logger("[*]","钱包余额:",self.wallet_balance)

    def Set_flag(self):
        import platform
        import os
        tmp_system = platform.system()
        filepath = ""
        if tmp_system == "Windows":
            filepath = "c:\\Windows\\Temp\\nekoflag"
        elif tmp_system == "Linux":
            filepath = "/tmp/nekoflag"
        elif tmp_system == "Darwin":
            filepath = "/tmp/nekoflag"
        os.system("echo run > "+filepath)

    def Clear_flag(self):
        import platform
        import os
        tmp_system = platform.system()
        filepath = ""
        if tmp_system == "Windows":
            filepath = "c:\\Windows\\Temp\\nekoflag"
        elif tmp_system == "Linux":
            filepath = "/tmp/nekoflag"
        elif tmp_system == "Darwin":
            filepath = "/tmp/nekoflag"
        os.system("echo close > "+filepath)

    def Check_flag(self):
        import platform
        tmp_system = platform.system()
        filepath = ""
        if tmp_system == "Windows":
            filepath = "c:\\Windows\\Temp\\nekoflag"
        elif tmp_system == "Linux":
            filepath = "/tmp/nekoflag"
        elif tmp_system == "Darwin":
            filepath = "/tmp/nekoflag"
        data = ""
        with open(filepath,"r") as f:
            data = f.read()
        if data == "run":
            return True
        else:
            return False

    def watch_dog_thread(self):
        import time
        import os
        import platform
        tmp_system = platform.system()
        if tmp_system == "Windows":
            command = "ping baidu.com -n 1 "
        elif tmp_system == "Linux":
            command = "ping baidu.com -t 1"
        elif tmp_system == "Darwin":
            command = "ping baidu.com -t 1"
        while True:
            try:
                p = os.popen(command)
                if "100%" in p.read():
                    self.Logger("[-]","网络状态异常,自动登陆")
                    self.login()
                else:
                    self.Logger("[+]","网络状态正常")
                time.sleep(60)
            except KeyboardInterrupt:
                self.Logger("[*]","手动退出")
                break

    def start(self):
        self.Set_flag()
        import multiprocessing
        import os
        self.Logger("[+]","开启守护进程")
        p = multiprocessing.Process(target=self.watch_dog_thread,daemon=False)
        p.start()
        self.Logger("[+]","守护进程启动成功")
    
    def stop(self):
        self.Logger("[+]","退出守护线程")
        self.Clear_flag()

def logo():
    print("""
====================================================================================== 
 __    __  ________  __    __   ______          ______   _______   __    __  __    __ 
/  \  /  |/        |/  |  /  | /      \        /      \ /       \ /  |  /  |/  \  /  |
$$  \ $$ |$$$$$$$$/ $$ | /$$/ /$$$$$$  |      /$$$$$$  |$$$$$$$  |$$ |  $$ |$$  \ $$ |
$$$  \$$ |$$ |__    $$ |/$$/  $$ |  $$ |      $$ \__$$/ $$ |__$$ |$$ |  $$ |$$$  \$$ |
$$$$  $$ |$$    |   $$  $$<   $$ |  $$ |      $$      \ $$    $$< $$ |  $$ |$$$$  $$ |
$$ $$ $$ |$$$$$/    $$$$$  \  $$ |  $$ |       $$$$$$  |$$$$$$$  |$$ |  $$ |$$ $$ $$ |
$$ |$$$$ |$$ |_____ $$ |$$  \ $$ \__$$ |      /  \__$$ |$$ |  $$ |$$ \__$$ |$$ |$$$$ |
$$ | $$$ |$$       |$$ | $$  |$$    $$/       $$    $$/ $$ |  $$ |$$    $$/ $$ | $$$ |
$$/   $$/ $$$$$$$$/ $$/   $$/  $$$$$$/         $$$$$$/  $$/   $$/  $$$$$$/  $$/   $$/ 
======================================================================================
|                                 深澜校园网认证程序                                 |
|                                 Power By Nekokami                                  |
+------------------------------------------------------------------------------------+
|   2020年11月16日                                                  NekoProject-10   |
+------------------------------------------------------------------------------------+

    """)

def get_opt():
    import argparse
    parser = argparse.ArgumentParser()
    parser.add_argument("-u", "--username", default=None,
                        help="登陆用户名(必填)")
    parser.add_argument("-p", "--password", default=None,
                        help="登陆密码(必填)")
    parser.add_argument("-s", "--server", default="192.168.118.51",
                        help="服务器地址(默认:192.168.118.51)")
    parser.add_argument("-c", "--cip", default="59.75.228.242",
                        help="客户端地址(默认:59.75.228.242)")
    parser.add_argument("-m", "--mode", type=int, default=0,
                        help="运行模式 0 仅登陆 1 前台模式")
    # parser.add_argument("-k", "--kill",default=False,
    #                     help="关闭守护进程", action="store_true")
    parser.add_argument("-v", "--version", default=False,
                        help="显示版本")
    args = parser.parse_args()
    if args.version:
        return False, args
    status = True
    if args.username is None or args.password is None:
        status = False
    return status, args


def main():
    logo()
    status,args = get_opt()
    if args.version:
        print("[*]","NekoSrun 深澜校园网认证程序")

    if not status and not args.version:
        print("[-] 请输入-h/--help 获取参数信息")
        return
    SLL = ShenLanLogin(username=args.username,password=args.password,password_base64="",server_ip=args.server,user_ip=args.cip)
    SLL.Logger("[*]","运行模式",("仅登陆" if args.mode == 0 else "常驻后台"))
    SLL.login()
    SLL.Show_User_Info()
    if args.mode == 0:
        pass
    elif args.mode == 1:
        SLL.watch_dog_thread()
    

if __name__ == "__main__":
    main()

下载链接

https://nekokami0527.com/resource/%E6%B7%B1%E6%BE%9C%E6%A0%A1%E5%9B%AD%E7%BD%91%E8%87%AA%E5%8A%A8%E7%99%BB%E5%BD%95%E8%84%9A%E6%9C%AC.zip