顯示具有 request 標籤的文章。 顯示所有文章
顯示具有 request 標籤的文章。 顯示所有文章

2025年4月22日 星期二

Python 開發筆記 - 使用 http.server 建置嵌入式產品網頁開發環境與 Proxy 機制

這個需求快十年前弄過,那時是用 webpack proxy server,可以在習慣的桌機開發網頁,接著對於 CGI 查詢就透過 proxy 導向到 device CGI 

然後,現在弄一個 Python ,方便後續在老 server 上運行,畢竟肥肥的程式碼搭配檔案系統大小寫問題,還是交給 server 檔案系統處理吧,如此,未來有部分 html/js/css code 要快速測試,就只需要到指定機器上,運行

$ python3 proxy-server.py --port 12345 --proxy-target 'http://192.168.123.234:80' --proxy-paths '/cgi-bin/' -d project/path/html/code/
Serving files from directory: project/path/html/code/
Serving HTTP on 0.0.0.0 port 12345 (multi-threaded) ...
Proxying paths ['/cgi-bin/'] to http://192.168.123.234:80
^C
Server stopped.

如此對於 CGI request 就會自動導向到 'http://192.168.123.234:80' 

程式碼純筆記,這都是 AI 產的:

#!/usr/bin/env python3

import http.server
import socketserver
import http.client
import argparse
import os
import sys
import socket
from urllib.parse import urlparse

# 使用 ThreadingMixIn 來處理並發請求
class ThreadedHTTPServer(socketserver.ThreadingMixIn, socketserver.TCPServer):
    daemon_threads = True
    allow_reuse_address = True


class ProxyHTTPRequestHandler(http.server.SimpleHTTPRequestHandler):
    # 代理目標地址
    proxy_target = None
    # 需要代理的路徑前綴
    proxy_paths = []
    # 連接池(簡單實現)
    conn_pool = {}
    
    def __init__(self, *args, **kwargs):
        # 確保可以正確處理目錄參數
        super().__init__(*args, **kwargs)
    
    def do_GET(self):
        # 檢查是否需要代理這個請求
        for path_prefix in self.proxy_paths:
            if self.path.startswith(path_prefix):
                self.proxy_request()
                return
        
        # 如果不需要代理,則使用默認處理方式
        super().do_GET()
    
    def do_POST(self):
        # 為 POST 請求也提供代理功能
        for path_prefix in self.proxy_paths:
            if self.path.startswith(path_prefix):
                self.proxy_request()
                return
        
        super().do_POST()
    
    def get_connection(self, host, is_ssl=False):
        """從連接池獲取連接,或建立新連接"""
        key = (host, is_ssl)
        
        if key not in self.conn_pool:
            if is_ssl:
                self.conn_pool[key] = http.client.HTTPSConnection(host)
            else:
                self.conn_pool[key] = http.client.HTTPConnection(host)
        
        # 檢查連接是否仍然有效
        conn = self.conn_pool[key]
        try:
            # 嘗試使用一個非阻塞的方式檢查連接狀態
            old_timeout = conn.sock.gettimeout()
            conn.sock.settimeout(0.01)
            conn.sock.recv(1, socket.MSG_PEEK)
            conn.sock.settimeout(old_timeout)
        except (socket.error, AttributeError):
            # 連接已關閉或存在問題,創建新連接
            if is_ssl:
                self.conn_pool[key] = http.client.HTTPSConnection(host)
            else:
                self.conn_pool[key] = http.client.HTTPConnection(host)
        except Exception:
            # 其他類型的錯誤,可能連接仍然有效
            pass
            
        return self.conn_pool[key]
    
    def proxy_request(self):
        """使用 http.client 處理代理請求,這在 Python 3.2 中更高效"""
        target_url = urlparse(self.proxy_target + self.path)
        
        # 獲取請求頭部
        headers = {}
        for key, value in self.headers.items():
            # 排除一些特定的頭部
            if key.lower() not in ('host', 'content-length'):
                headers[key] = value
        
        # 設置正確的 Host 頭部
        headers['Host'] = target_url.netloc
        
        # 讀取請求體(如果有)
        content_length = int(self.headers.get('Content-Length', 0))
        body = self.rfile.read(content_length) if content_length > 0 else None
        
        try:
            # 確定是否使用 HTTPS
            is_ssl = target_url.scheme == 'https'
            
            # 獲取或創建連接
            conn = self.get_connection(target_url.netloc, is_ssl)
            
            # 構建請求路徑
            request_path = target_url.path
            if target_url.query:
                request_path += '?' + target_url.query
            
            # 發送請求
            conn.request(
                method=self.command,
                url=request_path,
                body=body,
                headers=headers
            )
            
            # 獲取響應
            response = conn.getresponse()
            
            # 設置響應狀態碼
            self.send_response(response.status)
            
            # 設置響應頭部
            for header in response.getheaders():
                key, value = header
                if key.lower() != 'transfer-encoding':  # 排除特定頭部
                    self.send_header(key, value)
            self.end_headers()
            
            # 發送響應體
            self.wfile.write(response.read())
            
            # 不關閉連接,將其保留在連接池中
            
        except http.client.HTTPException as e:
            # 處理 HTTP 錯誤
            self.send_response(500)
            self.send_header('Content-Type', 'text/plain; charset=utf-8')
            self.end_headers()
            self.wfile.write("HTTP Error: {}".format(str(e)).encode('utf-8'))
            
        except Exception as e:
            # 處理其他錯誤
            self.send_response(500)
            self.send_header('Content-Type', 'text/plain; charset=utf-8')
            self.end_headers()
            self.wfile.write("Proxy error: {}".format(str(e)).encode('utf-8'))


def run_server(port=8000, proxy_target="http://localhost:8080", proxy_paths=["/cgi-bin/"], directory=None):
    # 設置代理目標和路徑
    ProxyHTTPRequestHandler.proxy_target = proxy_target
    ProxyHTTPRequestHandler.proxy_paths = proxy_paths
    
    # 確保代理目標是有效的 URL
    if not ProxyHTTPRequestHandler.proxy_target.startswith(('http://', 'https://')):
        ProxyHTTPRequestHandler.proxy_target = "http://{}".format(ProxyHTTPRequestHandler.proxy_target)
    
    # 移除代理目標末尾的斜線(如果有)
    if ProxyHTTPRequestHandler.proxy_target.endswith('/'):
        ProxyHTTPRequestHandler.proxy_target = ProxyHTTPRequestHandler.proxy_target[:-1]
    
    # 設置工作目錄
    if directory:
        os.chdir(directory)
        print("Serving files from directory: {}".format(directory))
    else:
        print("Serving files from current directory: {}".format(os.getcwd()))
    
    # 提高 socket 超時時間以處理慢速連接
    socket.setdefaulttimeout(60)
    
    # 建立服務器(使用線程化服務器)
    server_address = ("", port)
    httpd = ThreadedHTTPServer(server_address, ProxyHTTPRequestHandler)
    
    print("Serving HTTP on 0.0.0.0 port {} (multi-threaded) ...".format(port))
    print("Proxying paths {} to {}".format(proxy_paths, proxy_target))
    
    try:
        httpd.serve_forever()
    except KeyboardInterrupt:
        print("\nServer stopped.")
        httpd.server_close()


if __name__ == "__main__":
    # 解析命令行參數
    parser = argparse.ArgumentParser(description='HTTP Server with proxy capabilities (optimized for Python 3.2)')
    parser.add_argument('--port', type=int, default=8000, help='Port to listen on (default: 8000)')
    parser.add_argument('--proxy-target', type=str, default="http://localhost:8080", 
                        help='Target server to proxy to (default: http://localhost:8080)')
    parser.add_argument('--proxy-paths', type=str, default="/cgi-bin/", 
                        help='Comma-separated list of path prefixes to proxy (default: /cgi-bin/)')
    parser.add_argument('--directory', '-d', type=str, default=None,
                        help='Specify directory to serve files from (default: current directory)')
    
    args = parser.parse_args()
    proxy_paths = [path.strip() for path in args.proxy_paths.split(',')]
    
    run_server(args.port, args.proxy_target, proxy_paths, args.directory)

2021年6月12日 星期六

Node.js 筆記 - 使用 Puppeteer 監控瀏覽網頁的 Requests 跟 HTML Code @ macOS 11.4, node v14.14.0, npm 6.14.5

緣起於在追蹤網站架構時,每次都靠 chrome browser 開發人員工具去追蹤特定的 request 還好,若需要大量瀏覽找尋一些蛛絲馬跡時,就會很煩!以前很閒時,就會跳下去寫 chrome extension 或是熱衷於 C++ 時,就會寫一點點  Chromium Embedded Framework 來寫個小型瀏覽器,最後想起來試試看 Puppeteer 吧!

使用 Puppeteer 很輕鬆,除了是很常見的 Javascript 語言外,Puppeteer 套件提供了網頁端常見的事件偵測、所發的 Request 清單。就這樣拼拼湊湊即可完工:

(async () => {
const browser = await puppeteer.launch({headless: false});
//const page = await browser.newPage();
const context = await browser.createIncognitoBrowserContext();
const page = await context.newPage();
await page.emulate(device);
await page.setRequestInterception(true);

// https://pptr.dev/#?product=Puppeteer&version=v10.0.0&show=api-event-domcontentloaded
// https://developer.mozilla.org/en-US/docs/Web/API/Window/DOMContentLoaded_event
//
// The DOMContentLoaded event fires when the initial HTML document has been completely loaded and parsed, without waiting for stylesheets, images, and subframes to finish loading.
//
page.on('domcontentloaded', () => {
console.log('on.domcontentloaded');
});

// https://pptr.dev/#?product=Puppeteer&version=v10.0.0&show=api-event-domcontentloaded
// https://developer.mozilla.org/en-US/docs/Web/API/Window/load_event
//
// The load event is fired when the whole page has loaded, including all dependent resources such as stylesheets and images. This is in contrast to DOMContentLoaded, which is fired as soon as the page DOM has been loaded, without waiting for resources to finish loading.
//
page.on('load', async () => {
console.log('on.load');
watchTags(page, watch_tags);
});

// https://pptr.dev/#?product=Puppeteer&version=v10.0.0&show=api-event-framenavigated
// https://pptr.dev/#?product=Puppeteer&version=v10.0.0&show=api-class-frame
page.on('framenavigated', frame => {
console.log('on.framenavigated: '+frame.url());
});

// https://pptr.dev/#?product=Puppeteer&version=v10.0.0&show=api-event-request
// https://pptr.dev/#?product=Puppeteer&version=v10.0.0&show=api-class-httprequest
page.on('request', request => {
watchRequest(page.url(), request.url());
if (skip_resource_type[ request.resourceType() ])
request.abort();
else
request.continue();
});

// https://pptr.dev/#?product=Puppeteer&version=v10.0.0&show=api-event-response
// https://pptr.dev/#?product=Puppeteer&version=v10.0.0&show=api-class-httpresponse
page.on('response', response => {
//console.log('on.response: '+response.url());
});

while (true) {
if (url_init.length > 0) {
const target_url = url_init.shift();
await page.goto(target_url, {
waitUntil: 'networkidle0',
});
console.log('networkidle0');
}
if (url_init.length > 0)
await sleep(5000);
else
await sleep(50);
}
await browser.close();
})();

如此,搭配 command line 使用,就可很快的監控想要的 request 規則,或是 HTML DOM Tree 的變化,例如監控 <script> 的讀取位置在哪:

% node index.js -url "https://tw.yahoo.com" -tag "script.src"
...
on.load
[WATCH][TAG] script: src
Web URL: [https://tw.yahoo.com/]
[
  { src: 'https://s.yimg.com/wi/ytc.js' },
  {
    src: 'https://tw.mobi.yahoo.com/polyfill.min.js?features=array.isarray%2Carray.prototype.every%2Carray.prototype.foreach%2Carray.prototype.indexof%2Carray.prototype.map%2Cdate.now%2Cfunction.prototype.bind%2Cobject.keys%2Cstring.prototype.trim%2Cobject.defineproperty%2Cobject.defineproperties%2Cobject.create%2Cobject.freeze%2Carray.prototype.filter%2Carray.prototype.reduce%2Cobject.assign%2Cpromise%2Crequestanimationframe%2Carray.prototype.some%2Cobject.getownpropertynames%2Clocale-data-en-us%2Cintl%2Clocale-data-zh-hant-tw&version=2.1.23'
  },
  { src: 'https://s.yimg.com/aaq/yc/2.9.0/zh.js' },
  {
    src: 'https://s.yimg.com/ud/fp/js/vendor.174522d6d76b51858b93.min.js'
  },
  { src: 'https://s.yimg.com/ud/fp/js/common.js' },
  { src: 'https://s.yimg.com/oa/consent.js' },
  { src: 'https://consent.cmp.oath.com/cmpStub.min.js' },
  { src: 'https://consent.cmp.oath.com/cmp.js' },
  { src: 'https://s.yimg.com/ss/rapid3.js' },
  { src: 'https://s.yimg.com/rq/darla/boot.js' },
  {
    src: 'https://s.yimg.com/ud/fp/js/main.6622c6aeda051386afaa.min.js'
  },
  { src: 'https://mbp.yimg.com/sy/os/yaft/yaft-0.3.22.min.js' },
  {
    src: 'https://mbp.yimg.com/sy/os/yaft/yaft-plugin-aftnoad-0.1.3.min.js'
  },
  { src: 'https://s.yimg.com/aaq/vzm/perf-vitals_1.0.0.js' },
  {
    src: 'https://fc.yahoo.com/sdarla/php/client.php?dm=1&lang=zh-Hant-TW'
  },
  {
    src: 'https://s.yimg.com/aaq/c/e7fecc0.caas-abu_highlander.min.js'
  }
]
...

2021年1月14日 星期四

[Javascript] 複寫 XMLHttpRequest 來紀錄 Network Request URL @ Chrome Browser DevTools

最近 podcast 很夯,來研究 <audio> 。之前研究過 <video> 就有發現現在都用很夯 blob 的播放方式,透過 JS 片段下載資料交給 <video> 播放,以至於從 JS 調閱出 <video> 元件時,看不到真實的影片來源,而是一連串 blob (Binary Large Object) 記憶體位置。

原本想說在 Chrome DevTools 下,能不能靠 JS 取得 network request 發送清單來做應用,看著看著突然腦筋一轉,乾脆就用 XMLHttpRequest 好了,多包一層就可以收集了。

用法:

//
// https://stackoverflow.com/questions/7775767/javascript-overriding-xmlhttprequest-open
//
(function() {
var proxied = window.XMLHttpRequest.prototype.open;
window.XMLHttpRequest.prototype.open = function() {
// console.log( arguments );
//
// 只關注 m3u8 來源
//
if (arguments.length >= 2 && arguments[1].indexOf('.m3u8') > 0) {
console.log( arguments );
// return;
}
return proxied.apply(this, [].slice.call(arguments));
};
})();

2018年12月3日 星期一

Android 開發筆記 - Image Picker 與權限管理

在 Android 6.0 (API:23) 後,就算 AndroidManifest.xml 寫好要權限的部分,但實務上在進行時,依舊要主動要一次,讓用戶感受到真的要權限了:https://developer.android.com/training/permissions/requesting

做 Image Picker 時,若最終要讀取資料,那需要 android.permission.READ_EXTERNAL_STORAGE 權限,以下則是在 MainActivity 運行的範例:

<?xml version="1.0" encoding="utf-8"?>
<manifest xmlns:android="http://schemas.android.com/apk/res/android">

    <application
        android:allowBackup="true"
        android:icon="@mipmap/ic_launcher"
        android:label="@string/app_name"
        android:roundIcon="@mipmap/ic_launcher_round"
        android:supportsRtl="true"
        android:theme="@style/AppTheme">
        <activity android:name=".MainActivity">
            <intent-filter>
                <action android:name="android.intent.action.MAIN" />

                <category android:name="android.intent.category.LAUNCHER" />
            </intent-filter>
        </activity>
    </application>
    <uses-permission android:name="android.permission.READ_EXTERNAL_STORAGE"/>
</manifest>


public class MainActivity extends AppCompatActivity {
private static final int REQUEST_SELECT_VIDEO = 1;
private static final int REQUEST_EXTERNAL_STORAGE = 2;
private static String[] PERMISSIONS_STORAGE = {
Manifest.permission.READ_EXTERNAL_STORAGE,
};

@Override
protected void onCreate(Bundle savedInstanceState) {
super.onCreate(savedInstanceState);
setContentView(R.layout.activity_main);

if (ActivityCompat.checkSelfPermission(this, Manifest.permission.READ_EXTERNAL_STORAGE) != PackageManager.PERMISSION_GRANTED) {
ActivityCompat.requestPermissions(this, PERMISSIONS_STORAGE, REQUEST_EXTERNAL_STORAGE);
} else {
request_pick();
}
}

void request_pick() {
final Uri mUri = android.provider.MediaStore.Images.Media.EXTERNAL_CONTENT_URI;
final Intent mIntent = new Intent(Intent.ACTION_PICK, mUri);
final PackageManager mPackageManager = getPackageManager();
List<ResolveInfo> list = mPackageManager.queryIntentActivities(mIntent, PackageManager.MATCH_DEFAULT_ONLY);
if (list.size() > 1) {
startActivityForResult(
Intent.createChooser(
new Intent(Intent.ACTION_PICK, mUri), "選取圖片"
),
REQUEST_SELECT_VIDEO
);
} else {
startActivityForResult(mIntent, REQUEST_SELECT_VIDEO);
}
}

@Override
protected void onActivityResult(int requestCode, int resultCode, @Nullable Intent data) {
super.onActivityResult(requestCode, resultCode, data);
if (resultCode == RESULT_OK) {
if (requestCode == REQUEST_SELECT_VIDEO) {
// ...
// 在處理檔案讀取時,缺少 android.permission.READ_EXTERNAL_STORAGE 會造成 IOException:
// open failed: EACCES (Permission denied)
// ...
}
}
}
}