爬虫逆向（爬虫逆向进阶实战总结）#

1
from PIL import Image
2

3
def plus(str):
4
    return str.zfill(8)
5

6
def get_key(strr):
7
    str_ = ""
8
    for i in range(len(strr)):
9
        str_ = str_ + plus(bin(ord(strr[i])).replace('0b', ''))
10
    return str_
11

12
def mod(x, y):
13
    return x % y
14

15
def func(old_img, str2, new_img):
16
    im = Image.open(old_img)
17
    width = im.size[0]
18
    height = im.size[1]
19
    count = 0
20
    key = get_key(str2)
21
    keylen = len(key)
22
    for h in range(0, height):
23
        for w in range(0, width):
24
            pixel = im.getpixel((w, h))
25
            a = pixel[0]
26
            b = pixel[1]
27
            c = pixel[2]
28
            if count == keylen:
29
                break
30
            a = a - mod(a, 2) + int(key[count])
31
            count += 1
32
            if count == keylen:
33
                im.putpixel((w, h), (a, b, c))
34
                break
35
            b = b - mod(b, 2) + int(key[count])
36
            count += 1
37
            if count == keylen:
38
                im.putpixel((w, h), (a, b, c))
39
                break
40
            c = c - mod(c, 2) + int(key[count])
41
            count += 1
42
            if count == keylen:
43
                im.putpixel((w, h), (a, b, c))
44
                break
45
            if count % 3 == 0:
46
                im.putpixel((w, h), (a, b, c))
47
    im.save(new_img)
48

49
# 使用示例
50
old_img = r"timg.jpg"
51
new_img = r"timg2.jpg"
52
func(old_img, "Lx Is Good Man", new_img)

1
import base64
2
print(base64.b64encode('lx'.encode()))  # 编码
3
print(base64.b64decode('bHg='.encode()))  # 解码

1
import hashlib
2
m = hashlib.md5()
3
m.update(b'12')
4
print(m.hexdigest())  # 输出 202cb962ac59075b964b07152d234b70

1
import binascii
2
from pyDes import des, CBC, PAD_PKCS5
3

4
def des_encrypt(secret_key, s):
5
    iv = secret_key
6
    k = des(secret_key, CBC, iv, pad=None, padmode=PAD_PKCS5)
7
    en = k.encrypt(s, padmode=PAD_PKCS5)
8
    return binascii.b2a_hex(en)
9

10
def des_decrypt(secret_key, s):
11
    iv = secret_key
12
    k = des(secret_key, CBC, iv, pad=None, padmode=PAD_PKCS5)
13
    de = k.decrypt(binascii.a2b_hex(s), padmode=PAD_PKCS5)
14
    return de
15

16
secret_str = des_encrypt('999', 'lx-message')
17
clear_str = des_decrypt('999', secret_str)

1
import base64
2
from Crypto.Cipher import AES
3

4
def add_to_16(value):
5
    while len(value) % 16 != 0:
6
        value += '\0'
7
    return str.encode(value)
8

9
def encrypt(key, text):
10
    aes = AES.new(add_to_16(key), AES.MODE_ECB)
11
    encrypt_aes = aes.encrypt(add_to_16(text))
12
    encrypted_text = str(base64.encodebytes(encrypt_aes), encoding='utf-8')
13
    return encrypted_text
14

15
def decrypt(key, text):
16
    aes = AES.new(add_to_16(key), AES.MODE_ECB)
17
    base64_decrypted = base64.decodebytes(text.encode(encoding='utf-8'))
18
    decrypted_text = str(aes.decrypt(base64_decrypted), encoding='utf-8').replace('\0', '')
19
    return decrypted_text

1
let password = "lx123";
2
let key = "1234567890abcdef";
3
cfg = {
4
    mode: CryptoJs.mode.ECB,
5
    padding: CryptoJs.pad.Pkcs7
6
}
7
let encPwd = CryptoJs.AES.encrypt(password, key, cfg).toString();
8
let decPwd = CryptoJs.AES.decrypt(encPwd, key, cfg).toString(CryptoJs.enc.Utf8);

1
import base64
2
import rsa
3
from rsa import common
4

5
class RsaUtil(object):
6
    PUBLIC_KEY_PATH = 'public_key.pem'
7
    PRIVATE_KEY_PATH = 'private_key.pem'
8

9
    def __init__(self, company_pub_file=PUBLIC_KEY_PATH, company_pri_file=PRIVATE_KEY_PATH):
10
        if company_pub_file:
11
            self.company_public_key = rsa.PublicKey.load_pkcs1_openssl_pem(open(company_pub_file).read())
12
        if company_pri_file:
13
            self.company_private_key = rsa.PrivateKey.load_pkcs1(open(company_pri_file).read())
14

15
    def get_max_length(self, rsa_key, encrypt=True):
16
        blocksize = common.byte_size(rsa_key.n)
17
        reserve_size = 11
18
        if not encrypt:
19
            reserve_size = 0
20
        maxlength = blocksize - reserve_size
21
        return maxlength
22

23
    def encrypt_by_public_key(self, message):
24
        encrypt_result = b''
25
        max_length = self.get_max_length(self.company_public_key)
26
        while message:
27
            input = message[:max_length]
28
            message = message[max_length:]
29
            out = rsa.encrypt(input, self.company_public_key)
30
            encrypt_result += out
31
        encrypt_result = base64.b64encode(encrypt_result)
32
        return encrypt_result
33

34
    def decrypt_by_private_key(self, message):
35
        decrypt_result = b""
36
        max_length = self.get_max_length(self.company_private_key, False)
37
        decrypt_message = base64.b64decode(message)
38
        while decrypt_message:
39
            input = decrypt_message[:max_length]
40
            decrypt_message = decrypt_message[max_length:]
41
            out = rsa.decrypt(input, self.company_private_key)
42
            decrypt_result += out
43
        return decrypt_result

1
const JSEncrypt = require('jsencrypt');
2
publickey = '公钥';
3

4
let jse = new JSEncrypt();
5
jse.setPublicKey(publickey);
6
var encStr = jse.encrypt('username');
7

8
privatekey = '私钥';
9
jse.setPrivateKey(privatekey);
10
var Str = jse.decrypt(encStr);

1
import execjs
2
js = 'function test() { return "hello"; }'
3
xvt_anti = execjs.compile(js).call('test')

1
import hashlib
2
m = hashlib.md5()
3
m.update(b'12')
4
print(m.hexdigest())

1
from Cryptodome.PublicKey import RSA
2
from Cryptodome.Cipher import PKCS1_v1_5
3
import base64
4

5
def encrypt_str(data):
6
    key = "MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEAsgDq4OqxuEisnk2F0EJFmw4xKa5IrcqEYHvqxPs2CHEg2kolhfWA2SjNuGAHxyDDE5MLtOvzuXjBx/5YJtc9zj2xR/0moesS+Vi/xtG1tkVaTCba+TV+Y5C61iyr3FGqr+KOD4/XECu0Xky1W9ZmmaFADmZi7+6gO9wjgVpU9aLcBcw/loHOeJrCqjp7pA98hRJRY+MML8MK15mnC4ebooOva+mJlstW6t/1lghR8WNV8cocxgcHHuXBxgns2MlACQbSdJ8c6Z3RQeRZBzyjfey6JCCfbEKouVrWIUuPphBL3OANfgp0B+QG31bapvePTfXU48TYK0M5kE+8LgbbWQIDAQAB"
7
    rsakey = RSA.import_key(base64.b64decode(key))
8
    cipher = PKCS1_v1_5.new(rsakey)
9
    cipher_text = base64.b64encode(cipher.encrypt(data.encode(encoding="utf-8")))
10
    return cipher_text
11

12
password = encrypt_str("11")
13
print(password)

1
const { JSDOM } = require('jsdom');
2

3
const dom = new JSDOM(`
4
    <!DOCTYPE html>
5
    <html>
6
    <body>
7
        <script>
8
            // 模拟环境
9
            window = {
10
                document: {
11
                    location: {
12
                        href: 'https://example.com'
13
                    }
14
                }
15
            };
16
            // 你的JS代码
17
            function get_signature(url) {
18
                return "signature";
19
            }
20
        </script>
21
    </body>
22
    </html>
23
`);
24

25
const window = dom.window;
26
const document = window.document;
27

28
// 执行JS代码
29
const signature = window.get_signature('https://example.com');
30
console.log(signature);

1
# -*- coding: utf-8 -*-
2
import os
3
from selenium import webdriver
4

5
ua = 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/88.0.4324.190 Safari/537.36'
6
PRO_DIR = os.path.dirname(os.path.abspath(__file__))
7

8
s1 = """
9
    <!DOCTYPE html>
10
    <html style="font-size: 50px;">
11
    <head>
12
        <meta http-equiv="Content-Type" content="text/html; charset=UTF-8">
13
        <title>signature-hook</title>
14
    </head>
15
    <body></body>
16

17
    <script type="text/javascript">
18
    """
19
s2 = """
20
    </script>
21
    </html>
22
    """
23

24
def driver_sig(html_file):
25
    option = webdriver.ChromeOptions()
26
    option.add_argument('headless')
27
    option.add_argument('--no-sandbox')
28
    option.add_argument('--user-agent={}'.format(ua))
29
    driver = webdriver.Chrome(chrome_options=option)
30
    driver.get('file:///' + PRO_DIR + html_file)
31
    sig = driver.title
32
    driver.quit()
33
    return sig
34

35
sign_js = '''
36
window.navigator = {
37
    userAgent: '{}'
38
};
39
function get_sign() {
40
    // 模拟签名逻辑
41
    return "signature";
42
}
43
var signature = get_sign();
44
document.clear();
45
document.write(signature);
46
'''.format(ua)
47

48
if __name__ == '__main__':
49
    html_file = 'get_sign.html'
50
    with open(html_file, 'w', encoding='utf-8') as fw:
51
        fw.write(s1 + sign_js + s2)
52
    sig = driver_sig(html_file)
53
    print(sig)

1
window = new Proxy(global, {
2
    get: function(target, key, receiver) {
3
        console.log("window.get", key, target[key]);
4
        if (key === "location") {
5
            location = new Proxy(target[key], {
6
                get: function(_target, key, _receiver) {
7
                    console.log("window.get", key, _target[key]);
8
                    if (key === "port") {
9
                        console.log("公众号【Pythonxy】");
10
                    }
11
                    return _target[key];
12
                }
13
            });
14
        }
15
        return target[key];
16
    },
17
    set: function(target, key, value, receiver) {
18
        console.log("window.set", key, value);
19
        target[key] = value;
20
    }
21
});

1
from selenium.webdriver import Chrome
2
from selenium.webdriver import ChromeOptions
3

4
# 79年之前
5
option = ChromeOptions()
6
option.add_experimental_option('excludeSwitches', ['enable-automation'])
7
driver = Chrome(options=option)
8

9
# 79年之后
10
driver = Chrome()
11
driver.execute_cdp_cmd("Page.addScriptToEvaluateOnNewDocument", {
12
    "source": """Object.defineProperty(navigator, 'webdriver', {get: () => false})"""
13
})
14

15
# 88版本之后
16
chrome_options = ChromeOptions()
17
chrome_options.add_argument("disable-blink-features")
18
chrome_options.add_argument("disable-blink-features=AutomationControlled")
19
driver = Chrome(options=chrome_options)

1
ws.onmessage = function(event) {
2
  var data = event.data;
3
};

1
import asyncio
2
import websockets
3

4
async def check_permit(websocket):
5
    send_text = 'lx'
6
    await websocket.send(send_text)
7
    return True
8

9
async def recv_msg(websocket):
10
    while 1:
11
        recv_text = await websocket.recv()
12
        print(recv_text)
13

14
async def main_logic(websocket, path):
15
    await check_permit(websocket)
16
    await recv_msg(websocket)
17

18
start_server = websockets.serve(main_logic, '127.0.0.1', 9999)
19
asyncio.get_event_loop().run_until_complete(start_server)
20
asyncio.get_event_loop().run_forever()

1
import websocket
2
import json
3

4
def on_message(ws, message):
5
    print("Received: " + message)
6

7
def on_error(ws, error):
8
    print("Error: " + str(error))
9

10
def on_close(ws, close_status_code, close_msg):
11
    print("Connection closed")
12

13
def on_open(ws):
14
    ws.send(json.dumps({"type": "login", "data": {"username": "user", "password": "pass"}}))
15

16
if __name__ == "__main__":
17
    ws = websocket.WebSocketApp("wss://example.com/socket",
18
                              on_open=on_open,
19
                              on_message=on_message,
20
                              on_error=on_error,
21
                              on_close=on_close)
22
    ws.run_forever()