GVSDS/Viper_VRTE
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

'基线重置，修复了一些 bug，fix(llvmlite): build_alloca_at_entry 在已终止入口块后在 br 后追加 alloca 致非法 IR嵌套循环（如 LoopTest.test_nested_loop 内层 for）内联新建循环变量 alloca 时，入口块已以 br 终止，block_append_text 把 alloca 追加到终止指令之后，生成不可达指令，llc 报 "instruction expected to be numbered %N or greater"。改用 block_insert_text_before_terminator：未终止等同追加，已终止插到终止指令之前；编号乱序由 RenumberSSA 重排处理。'

Browse Source
This commit is contained in:
TermiNexus committed 2026-09-07 15:05:50 +08:00
1 parent 62bceb6930
commit f53c2216aa
11 files changed
+2139 -117
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

requests/__init__.vp
+898
View File
@@ -0,0 +1,898 @@
import t, c
from stdint import *
import string
import socket
import memhub
import stdio
import stdlib
import zlib.pyzlib as pyzlib
# ============================================================
# requests/__init__.vp — HTTP(S) 客户端库包入口（Python requests 风格）
#
# 包结构:
# requests HTTP 客户端（纯 socket 实现，本文件）
# requests.https HTTPS 客户端（openssl SSL/TLS + 证书核验，见 https.vp）
#
# Usage:
# requests._mbuddy = my_mbuddy
# resp = requests.get("http://example.com/path")
# if resp is not None:
# print(resp.status_code)
# print(resp.text)
# requests.release(resp)
# ============================================================
# Module-level global MBuddy pointer (assigned by the user in main)
_mbuddy: memhub.MemBuddy | t.CPtr
# Socket initialization flag
_socket_ready: bool = False
HTTP_PORT: t.CDefine = 80
HTTP_BUFSIZ: t.CDefine = 4096
GZIP_BUFSIZ: t.CDefine = 65536
# 模块内重定向 / 失败诊断（静态缓冲）
_http_err_store: t.CArray[t.CChar, 512]
def last_error() -> str:
"""返回 HTTP 层最近一次失败的原因（无则空串）。"""
return _http_err_store
def _SetHttpError(text: str):
if text is None: return
_http_err_store[0] = 0
stdio.snprintf(_http_err_store, 512, "%s", text)
# ============================================================
# Response — HTTP response object
# ============================================================
class Response:
"""HTTP response object, similar to how Python's requests.Response works.
Allocated from _mbuddy, call release(resp) to free it after use.
"""
status_code: INT # HTTP status code (e.g., 200, 404)
text: str # Response body text
body_len: INT # Response body length
_raw_buf: str # 原始响应缓冲（含头与正文，release 释放）
_body_owned: str # 独立解码正文缓冲（chunked 等解码结果，release 一并释放）
# ============================================================
# Internal auxiliary functions
# ============================================================
def _EnsureSocketInit() -> INT:
"""Ensure Socket library is initialized (idempotent)."""
global _socket_ready
if _socket_ready: return 0
result: INT = socket.SocketInit()
if result == 0:
_socket_ready = True
return result
def _AllocResponse() -> Response | t.CPtr:
"""Allocate Response struct from _mbuddy and initialize it in-place."""
if _mbuddy is None: return None
resp: Response | t.CPtr = _mbuddy.alloc(Response.__sizeof__())
if resp is None: return None
resp.status_code = 0
resp.text = None
resp.body_len = 0
resp._raw_buf = None
resp._body_owned = None
return resp
def _ParseUrl(url: str, host_buf: str, host_size: INT, path_buf: str, path_size: INT, port: INT | t.CPtr) -> INT:
"""Parse URL and extract host, port, and path."""
if url is None: return -1
if host_buf is None or path_buf is None or port is None: return -1
p: str = url
# Skip "http://"
if string.strncmp(p, "http://", 7) == 0:
p += 7
# Parse host
host_start: str = p
host_len: INT = 0
while p[0] != '\0' and p[0] != ':' and p[0] != '/':
host_len += 1
p += 1
if host_len == 0 or host_len >= host_size: return -1
string.strncpy(host_buf, host_start, host_len)
host_buf[host_len] = 0
# Parse port
if p[0] == ':':
p += 1
port_val: INT = 0
while p[0] >= '0' and p[0] <= '9':
port_val = port_val * 10 + INT(p[0] - '0')
p += 1
c.DerefAs(port, port_val)
else:
c.DerefAs(port, HTTP_PORT)
# Parse path
if p[0] == '/':
path_start: str = p
path_len: INT = 0
while p[0] != '\0':
path_len += 1
p += 1
if path_len >= path_size: return -1
string.strncpy(path_buf, path_start, path_len)
path_buf[path_len] = 0
else:
if path_size < 2: return -1
path_buf[0] = '/'
path_buf[1] = 0
return 0
def _ParseStatusLine(resp_text: str, out_code: INT | t.CPtr) -> INT:
"""Parse HTTP status line and extract status code."""
if resp_text is None or out_code is None: return -1
p: str = resp_text
# Skip "HTTP/x.x "
while p[0] != '\0' and p[0] != ' ':
p += 1
if p[0] == ' ': p += 1
# Parse status code
code: INT = 0
while p[0] >= '0' and p[0] <= '9':
code = code * 10 + INT(p[0] - '0')
p += 1
c.DerefAs(out_code, code)
# Find \n
while p[0] != '\0' and p[0] != '\n':
p += 1
if p[0] == '\n': p += 1
return INT(t.CUInt64T(p) - t.CUInt64T(resp_text))
def _FindBodyStart(resp_text: str, resp_len: INT) -> INT:
"""Find the start of the HTTP response body (after rnrn)."""
if resp_text is None: return -1
end_of_headers: str = string.strstr(resp_text, "\r\n\r\n")
if end_of_headers is None:
end_of_headers = string.strstr(resp_text, "\n\n")
if end_of_headers is None:
return -1
return INT(t.CUInt64T(end_of_headers) - t.CUInt64T(resp_text)) + 2
return INT(t.CUInt64T(end_of_headers) - t.CUInt64T(resp_text)) + 4
# ============================================================
# 公共辅助（供 requests.https 等子模块复用）
# ============================================================
def create_response() -> Response | t.CPtr:
"""Create and initialize an empty HTTP Response (allocated from _mbuddy).
Call release(resp) to free it after use.
"""
return _AllocResponse()
def parse_status_line(resp_text: str, out_code: INT | t.CPtr) -> INT:
"""Parse HTTP status line, write status code into out_code."""
return _ParseStatusLine(resp_text, out_code)
def find_body_start(resp_text: str, resp_len: INT) -> INT:
"""Locate the start of the HTTP response body."""
return _FindBodyStart(resp_text, resp_len)
# ============================================================
# 响应解析（header 查询 / chunked 解码 / 统一 parse_response）
# ============================================================
def _CIEq(a: t.CChar, b: t.CChar) -> bool:
"""大小写不敏感字符相等（仅 ASCII）。"""
if a >= 'A' and a <= 'Z': a = t.CChar(INT(a) - INT('A') + INT('a'))
if b >= 'A' and b <= 'Z': b = t.CChar(INT(b) - INT('A') + INT('a'))
return a == b
def header_value(resp: Response | t.CPtr, name: str, out: str, out_size: INT) -> INT:
"""在原始响应中按名查询响应头（大小写不敏感，返回首值）。
Args:
resp: Response（其 _raw_buf 保留原始响应）
name: 头名称，如 "Content-Type"
out / out_size: 值写入缓冲
Returns:
0 找到并拷贝；-1 未找到或参数非法。
"""
if resp is None or resp._raw_buf is None or name is None or out is None: return -1
p: str = resp._raw_buf
# 跳过状态行
while p[0] != '\0' and p[0] != '\n':
p += 1
if p[0] == '\n': p += 1
# 逐头行匹配
while p[0] != '\0' and p[0] != '\r' and p[0] != '\n':
h: str = p
q: str = name
matched: bool = True
while q[0] != '\0':
if h[0] != q[0] and not _CIEq(h[0], q[0]):
matched = False
break
h += 1
q += 1
# name 匹配后允许 [空格]* ':'（RFC 无空格，宽松兼容）
while h[0] == ' ':
h += 1
if matched and h[0] == ':':
h += 1
while h[0] == ' ' or h[0] == '\t':
h += 1
i: INT = 0
while h[0] != '\0' and h[0] != '\r' and h[0] != '\n' and i < out_size - 1:
out[i] = h[0]
i += 1
h += 1
out[i] = 0
return 0
# 跳到下一行
while p[0] != '\0' and p[0] != '\n':
p += 1
if p[0] == '\n': p += 1
return -1
def _HexVal(ch: t.CChar) -> INT:
if ch >= '0' and ch <= '9': return INT(ch - '0')
if ch >= 'a' and ch <= 'f': return INT(ch - 'a') + 10
if ch >= 'A' and ch <= 'F': return INT(ch - 'A') + 10
return -1
def decode_chunked(src: str, src_len: INT, dst: str, dst_cap: INT) -> INT:
"""按 RFC 7230 解码 chunked 正文（不处理 chunk-ext，trailer 忽略）。
Args:
src / src_len: chunked 编码数据
dst / dst_cap: 解码输出缓冲
Returns:
解码字节数；失败返回 -1。
"""
if src is None or dst is None: return -1
r: INT = 0
w: INT = 0
while r < src_len:
# 1) chunk size（hex，可带 ;ext 扩展）
size: INT = 0
digit_seen: bool = False
while r < src_len:
ch: t.CChar = src[r]
if ch == '\r' or ch == '\n' or ch == ';': break
hv: INT = _HexVal(ch)
if hv < 0: return -1
digit_seen = True
if size > 0x7fffffff: return -1
size = size * 16 + hv
r += 1
if not digit_seen: return -1
# 2) 跳过扩展与行尾
while r < src_len and src[r] != '\n':
r += 1
if r < src_len and src[r] == '\n': r += 1
# 3) 结束块
if size == 0: return w
# 4) 拷贝数据 + 跳过 CRLF
if w + size > dst_cap: return -1
k: INT = 0
while k < size:
dst[w + k] = src[r + k]
k += 1
w += size
r += size
if r < src_len and src[r] == '\r': r += 1
if r < src_len and src[r] == '\n': r += 1
return w
def parse_response(resp: Response | t.CPtr, total: INT) -> INT:
"""统一解析已接收的原始响应并填充 Response 字段。
- 解析状态码 / 定位正文
- Transfer-Encoding: chunked 时解码到独立缓冲（_body_owned），
原始缓冲 _raw_buf 保留（供 header_value 查询）
Returns:
0 成功；-1 失败（此时 text=None / body_len=0）。
"""
if resp is None or resp._raw_buf is None: return -1
buf: str = resp._raw_buf
buf[total] = 0
_ParseStatusLine(buf, c.Addr(resp.status_code))
body_offset: INT = _FindBodyStart(buf, total)
if body_offset < 0 or body_offset >= total:
resp.text = None
resp.body_len = 0
return 0
# 检测 Transfer-Encoding: chunked
te_buf: t.CArray[t.CChar, 64]
chunked: bool = False
if header_value(resp, "Transfer-Encoding", te_buf, 64) == 0:
if string.strstr(te_buf, "chunked") is not None:
chunked = True
if chunked:
new_buf: str = _mbuddy.alloc(HTTP_BUFSIZ)
if new_buf is None:
resp.text = None
resp.body_len = 0
return -1
dec_len: INT = decode_chunked(buf + body_offset, total - body_offset, new_buf, HTTP_BUFSIZ)
if dec_len < 0:
_mbuddy.free(new_buf)
resp.text = None
resp.body_len = 0
return -1
new_buf[dec_len] = 0
resp._body_owned = new_buf
resp.text = new_buf
resp.body_len = dec_len
else:
resp.text = buf + body_offset
resp.body_len = total - body_offset
# ---- Content-Encoding: gzip 解压（zlib.pyzlib，纯 Viper 实现）----
ce_buf: t.CArray[t.CChar, 64]
if header_value(resp, "Content-Encoding", ce_buf, 64) == 0 \
and string.strstr(ce_buf, "gzip") is not None \
and resp.text is not None and resp.body_len > 0:
src_len2: t.CSizeT = t.CSizeT(resp.body_len)
olen2: t.CSizeT = 0
gz_out: BYTEPTR = pyzlib.decompress(_mbuddy, BYTEPTR(resp.text), src_len2, 15, GZIP_BUFSIZ, c.Addr(olen2))
if gz_out is not None:
old_owned: str = resp._body_owned
if old_owned is not None:
_mbuddy.free(old_owned)
gz_str: str = t.CChar(t.CUInt64T(gz_out), t.CPtr)
gz_str[olen2] = 0
resp._body_owned = gz_str
resp.text = gz_str
resp.body_len = INT(olen2)
else:
# 解压失败：保留原始 body，记录诊断信息
stdio.snprintf(_http_err_store, 512, "gzip decode failed: %s", pyzlib.get_error())
return 0
# ============================================================
# 公共 API
# ============================================================
def _HttpGetOnce(url: str) -> Response | t.CPtr:
"""HTTP GET 单次请求（不跟随重定向）。请求头带 Accept-Encoding: gzip。"""
if _mbuddy is None: return None
resp: Response | t.CPtr = _AllocResponse()
if resp is None: return None
if _EnsureSocketInit() != 0:
release(resp)
return None
# Parse URL
host_buf: t.CArray[t.CChar, 256]
path_buf: t.CArray[t.CChar, 1024]
port: INT = 0
if _ParseUrl(url, host_buf, 256, path_buf, 1024, c.Addr(port)) != 0:
release(resp)
return None
# Create Socket
sock = socket.Socket(socket.AF_INET, socket.SOCK_STREAM)
if sock.fd == socket.INVALID_SOCKET:
release(resp)
return None
if sock.connect(host_buf, port) != 0:
sock.close()
release(resp)
return None
# Build HTTP request
req_buf: str = _mbuddy.alloc(HTTP_BUFSIZ)
if req_buf is None:
sock.close()
release(resp)
return None
stdio.snprintf(req_buf, HTTP_BUFSIZ,
"GET %s HTTP/1.1\r\nHost: %s\r\nAccept-Encoding: gzip\r\nConnection: close\r\n\r\n",
path_buf, host_buf)
req_len: INT = INT(string.strlen(req_buf))
sent: INT = sock.send(req_buf, req_len)
_mbuddy.free(req_buf)
if sent <= 0:
sock.close()
release(resp)
return None
# Receive response body
resp_buf: str = _mbuddy.alloc(HTTP_BUFSIZ)
if resp_buf is None:
sock.close()
release(resp)
return None
total: INT = 0
while total < HTTP_BUFSIZ - 1:
n: INT = sock.recv(resp_buf + total, HTTP_BUFSIZ - 1 - total)
if n <= 0: break
total += n
resp_buf[total] = 0
sock.close()
resp._raw_buf = resp_buf
parse_response(resp, total)
return resp
def _HttpPostOnce(url: str, data: str = None, content_type: str = "application/x-www-form-urlencoded") -> Response | t.CPtr:
"""HTTP POST 单次请求（不跟随重定向）。请求头带 Accept-Encoding: gzip。"""
if _mbuddy is None: return None
resp: Response | t.CPtr = _AllocResponse()
if resp is None: return None
if _EnsureSocketInit() != 0:
release(resp)
return None
# Parse URL
host_buf: t.CArray[t.CChar, 256]
path_buf: t.CArray[t.CChar, 1024]
port: INT = 0
if _ParseUrl(url, host_buf, 256, path_buf, 1024, c.Addr(port)) != 0:
release(resp)
return None
body_len: INT = INT(string.strlen(data)) if data is not None else 0
ct: str = content_type if content_type is not None else "application/x-www-form-urlencoded"
# Create Socket
sock = socket.Socket(socket.AF_INET, socket.SOCK_STREAM)
if sock.fd == socket.INVALID_SOCKET:
release(resp)
return None
if sock.connect(host_buf, port) != 0:
sock.close()
release(resp)
return None
# Build HTTP request
req_buf: str = _mbuddy.alloc(HTTP_BUFSIZ)
if req_buf is None:
sock.close()
release(resp)
return None
stdio.snprintf(req_buf, HTTP_BUFSIZ,
"POST %s HTTP/1.1\r\nHost: %s\r\nContent-Type: %s\r\nContent-Length: %d\r\nAccept-Encoding: gzip\r\nConnection: close\r\n\r\n",
path_buf, host_buf, ct, body_len)
hdr_len: INT = INT(string.strlen(req_buf))
# Send headers + body
sent: INT = sock.send(req_buf, hdr_len)
if sent > 0 and body_len > 0 and data is not None:
sock.send(data, body_len)
_mbuddy.free(req_buf)
# Receive response body
resp_buf: str = _mbuddy.alloc(HTTP_BUFSIZ)
if resp_buf is None:
sock.close()
release(resp)
return None
total: INT = 0
while total < HTTP_BUFSIZ - 1:
n: INT = sock.recv(resp_buf + total, HTTP_BUFSIZ - 1 - total)
if n <= 0: break
total += n
resp_buf[total] = 0
sock.close()
resp._raw_buf = resp_buf
parse_response(resp, total)
return resp
# ============================================================
# 重定向跟随
# ============================================================
def _ResolveRedirectUrl(base: str, loc: str, out: str, cap: INT) -> INT:
"""把 Location 合并为完整 URL（支持绝对 / 协议相对 / 根相对），结果写 out。
Returns: 0 成功；-1 失败。
"""
if base is None or loc is None or out is None: return -1
lb: INT = INT(string.strlen(loc))
if string.strncmp(loc, "http://", 7) == 0 or string.strncmp(loc, "https://", 8) == 0:
if lb >= cap: return -1
string.strcpy(out, loc)
return 0
# 取 base 的 scheme://authority 前缀
q: str = base
while q[0] != '\0' and q[0] != ':': q += 1
if q[0] != ':': return -1
q += 1
if q[0] != '/': return -1
q += 1
if q[0] != '/': return -1
q += 1
while q[0] != '\0' and q[0] != '/': q += 1
prefix_len: INT = INT(t.CUInt64T(q) - t.CUInt64T(base))
need: INT = prefix_len + lb + 2
if need >= cap: return -1
i: INT = 0
while i < prefix_len:
out[i] = base[i]
i += 1
if loc[0] == '/':
j: INT = 0
while loc[j] != '\0':
out[prefix_len + j] = loc[j]
j += 1
out[prefix_len + j] = 0
else:
out[prefix_len] = '/'
j: INT = 0
while loc[j] != '\0':
out[prefix_len + 1 + j] = loc[j]
j += 1
out[prefix_len + 1 + j] = 0
return 0
def _HttpRedirectLoop(method: str, url: str, data: str, content_type: str) -> Response | t.CPtr:
"""GET/POST + 重定向跟随（http:// 链，最多 5 跳）。"""
if _mbuddy is None: return None
cur: str = _mbuddy.alloc(2048)
if cur is None: return None
if INT(string.strlen(url)) >= 2047:
_SetHttpError("url too long")
_mbuddy.free(cur)
return None
string.strcpy(cur, url)
hop: INT = 0
while hop < 5:
resp: Response | t.CPtr = None
if string.strcmp(method, "POST") == 0:
resp = _HttpPostOnce(cur, data, content_type)
else:
resp = _HttpGetOnce(cur)
if resp is None:
_mbuddy.free(cur)
return None
code: INT = resp.status_code
if not _IsHttpRedirect(code):
_mbuddy.free(cur)
return resp
loc: str = _mbuddy.alloc(2048)
if loc is None:
_mbuddy.free(cur)
return resp
if header_value(resp, "Location", loc, 2048) != 0:
_mbuddy.free(loc)
_mbuddy.free(cur)
return resp
nxt: str = _mbuddy.alloc(2048)
if nxt is None:
_mbuddy.free(loc)
_mbuddy.free(cur)
return resp
if _ResolveRedirectUrl(cur, loc, nxt, 2048) != 0:
_SetHttpError("unresolvable redirect Location")
release(resp)
_mbuddy.free(nxt)
_mbuddy.free(loc)
_mbuddy.free(cur)
return None
if string.strncmp(nxt, "https://", 8) == 0:
_SetHttpError("redirect to https:// needs requests.https")
release(resp)
_mbuddy.free(nxt)
_mbuddy.free(loc)
_mbuddy.free(cur)
return None
release(resp)
_mbuddy.free(loc)
_mbuddy.free(cur)
cur = nxt
hop += 1
_SetHttpError("too many redirects")
_mbuddy.free(cur)
return None
def _IsHttpRedirect(code: INT) -> bool:
return code >= 300 and code < 400
def get(url: str) -> Response | t.CPtr:
"""HTTP GET（自动跟随 3xx 重定向，最多 5 跳）。
Args:
url: Full URL (e.g., "http://example.com/path")
Returns:
Pointer to a Response object, None on failure (see last_error()).
"""
return _HttpRedirectLoop("GET", url, None, None)
def post(url: str, data: str = None, content_type: str = "application/x-www-form-urlencoded") -> Response | t.CPtr:
"""HTTP POST（自动跟随 3xx 重定向，最多 5 跳）。"""
return _HttpRedirectLoop("POST", url, data, content_type)
def release(resp: Response | t.CPtr):
"""Release Response object and its internal buffers.
Args:
resp: The Response object to release
"""
if resp is None: return
if _mbuddy is None: return
if resp._body_owned is not None:
_mbuddy.free(resp._body_owned)
if resp._raw_buf is not None:
_mbuddy.free(resp._raw_buf)
_mbuddy.free(t.CVoid(t.CUInt64T(resp), t.CPtr))
# ============================================================
# 大文件 Range 下载原语（requests.download 使用）
#
# _DlOnce 单次发起 "Range: bytes=start-end" 的 GET（不带 Accept-Encoding，
# 保证响应是原始字节），把响应体写入调用方提供的 dst 缓冲。
# 线程安全：本函数不使用 _mbuddy（响应/请求缓冲一律 stdlib），
# socket 连接独立，供多线程并行切片下载并发调用。
#
# 返回: >=0 响应体字节数；-1 参数错；-2 3xx 重定向（Location 写 out_loc，
# 由 download 模块跟随）；-3 其它失败（last_error() 查看原因）。
# out_status/out_total 分别输出 HTTP 状态码与 Content-Range 总大小。
# ============================================================
def _RawHeaderValue(buf: str, total: INT, name: str, out: str, out_size: INT) -> INT:
"""对原始响应头区做大小写不敏感查询（0=命中并拷贝值；-1=未命中）。"""
if buf is None or name is None or out is None or out_size <= 0:
return -1
body: INT = _FindBodyStart(buf, total)
if body < 0:
body = total
nl: INT = 0
while nl < body:
eol: INT = nl
while eol < body and buf[eol] != '\r' and buf[eol] != '\n':
eol += 1
if eol > nl:
k: INT = 0
ok: bool = True
while name[k] != 0 and nl + k < eol:
if not _CIEq(buf[nl + k], name[k]):
ok = False
break
k += 1
if ok and name[k] == 0 and nl + k < eol and buf[nl + k] == ':':
v: INT = nl + k + 1
while v < eol and (buf[v] == ' ' or buf[v] == '\t'):
v += 1
vlen: INT = eol - v
if vlen > out_size - 1:
vlen = out_size - 1
for i in range(vlen):
out[i] = buf[v + i]
out[vlen] = 0
return 0
nl = eol
while nl < body and (buf[nl] == '\r' or buf[nl] == '\n'):
nl += 1
return -1
def _StatusRaw(buf: str) -> INT:
"""从 "HTTP/1.1 206 ..." 首行解析状态码。"""
if buf is None:
return 0
i: INT = 0
while buf[i] == ' ':
i += 1
while buf[i] != 0 and buf[i] != ' ':
i += 1
while buf[i] == ' ':
i += 1
code: INT = 0
while buf[i] >= '0' and buf[i] <= '9':
code = code * 10 + INT(buf[i] - '0')
i += 1
return code
def _CLenOf(buf: str, total: INT) -> INT:
"""读取 Content-Length（无则 -1）。"""
hbuf: t.CArray[t.CChar, 32]
if _RawHeaderValue(buf, total, "Content-Length", hbuf, 32) != 0:
return -1
v: INT = 0
i: INT = 0
while hbuf[i] >= '0' and hbuf[i] <= '9':
v = v * 10 + INT(hbuf[i] - '0')
i += 1
return v
def _CRTotalOf(buf: str, total: INT) -> t.CSizeT:
"""从 Content-Range: bytes s-e/TOTAL 解析 TOTAL（非 206/缺头则 0）。"""
hbuf: t.CArray[t.CChar, 128]
if _RawHeaderValue(buf, total, "Content-Range", hbuf, 128) != 0:
return 0
p: INT = 0
while hbuf[p] != 0 and hbuf[p] != '/':
p += 1
if hbuf[p] == 0:
return 0
p += 1
v: t.CSizeT = 0
while hbuf[p] != 0 and hbuf[p] >= '0' and hbuf[p] <= '9':
v = v * 10 + t.CSizeT(hbuf[p] - '0')
p += 1
return v
def _DlOnce(url: str, start: t.CSizeT, count: INT, dst: str, dst_cap: INT,
out_status: INT | t.CPtr, out_loc: str, out_loccap: INT,
out_total: t.CSizeT | t.CPtr) -> INT:
"""HTTP 单次 Range 拉取（线程安全，见模块注释）。"""
if url is None or dst is None or dst_cap <= 0 \
or out_status is None or out_total is None:
return -1
c.Set(c.Deref(out_status), 0)
c.Set(c.Deref(out_total), 0)
if out_loc is not None and out_loccap > 0:
out_loc[0] = 0
if _EnsureSocketInit() != 0:
_SetHttpError("socket init failed")
return -3
host_buf: t.CArray[t.CChar, 256]
path_buf: t.CArray[t.CChar, 1024]
port: INT = 0
if _ParseUrl(url, host_buf, 256, path_buf, 1024, c.Addr(port)) != 0:
_SetHttpError("bad http url")
return -3
sock = socket.Socket(socket.AF_INET, socket.SOCK_STREAM)
if sock.fd == socket.INVALID_SOCKET:
_SetHttpError("socket() failed")
return -3
if sock.connect(host_buf, port) != 0:
_SetHttpError("connect failed")
sock.close()
return -3
req: str = stdlib.malloc(1024)
if req is None:
sock.close()
_SetHttpError("out of memory")
return -3
cnt: INT = count
if cnt < 0:
cnt = 0
end: t.CSizeT = start + t.CSizeT(cnt) - 1
if cnt <= 0:
end = start
stdio.snprintf(req, 1024,
"GET %s HTTP/1.1\r\nHost: %s\r\nRange: bytes=%zu-%zu\r\nConnection: close\r\n\r\n",
path_buf, host_buf, start, end)
sent: INT = sock.send(req, INT(string.strlen(req)))
stdlib.free(req)
if sent <= 0:
sock.close()
_SetHttpError("send failed")
return -3
total: INT = 0
tgt: INT = -1
while total < dst_cap - 1:
room: INT = dst_cap - 1 - total
if room > 4096:
room = 4096
n: INT = sock.recv(dst + total, room)
if n <= 0:
break
total += n
if tgt < 0:
bo: INT = _FindBodyStart(dst, total)
if bo >= 0:
cl: INT = _CLenOf(dst, total)
if cl >= 0:
tgt = bo + cl
if tgt >= 0 and total >= tgt:
break
sock.close()
dst[total] = 0
st: INT = _StatusRaw(dst)
c.Set(c.Deref(out_status), st)
if st == 0:
_SetHttpError("empty response")
return -3
if _IsHttpRedirect(st):
if out_loc is not None and out_loccap > 0:
if _RawHeaderValue(dst, total, "Location", out_loc, out_loccap) != 0:
out_loc[0] = 0
return -2
if st != 200 and st != 206:
stdio.snprintf(_http_err_store, 512, "http status %d", st)
return -3
if st == 206:
c.Set(c.Deref(out_total), _CRTotalOf(dst, total))
else:
cl2: INT = _CLenOf(dst, total)
if cl2 > 0:
c.Set(c.Deref(out_total), t.CSizeT(cl2))
bo: INT = _FindBodyStart(dst, total)
if bo < 0 or bo >= total:
return 0
blen: INT = total - bo
if blen > cnt and cnt > 0:
blen = cnt
return blen
requests/download.vp
+436
View File
@@ -0,0 +1,436 @@
import t, c
from stdint import *
import string
import stdio
import stdlib
import requests
import requests.https as rhttps
import vthreading as vt
import w32.win32file as wf
import w32.win32base as wb
import w32.win32sync as wsync
# ============================================================
# requests/download.vp — 大文件切片下载（Range + 串行/多线程）
#
# 用法:
# import requests.download as rd
# rc: INT = rd.download("https://host/big.bin", "big.bin", threads=4,
# chunk_size=262144, verify=True, ca_file="ca.crt")
# if rc != 0:
# print(rd.last_error())
#
# 策略:
# threads == 1 串行 Range 切片（顺序拉取并写入文件）
# threads > 1 多线程：文件按线程数均分为若干带(band)，每线程一个
# 连接顺序下载自己带的切片，写入同一文件的对应偏移。
#
# 说明:
# - 通过 Content-Range 探测总大小与服务器 Range 支持；
# - http:// 走 requests（纯 socket），https:// 走 requests.https（openssl）；
# - 每片请求不带 Accept-Encoding，响应为原始字节，直接落盘。
# ============================================================
DL_MAX_THREADS: t.CDefine = 8
DL_DEFAULT_CHUNK: t.CDefine = 262144
DL_REDIRECT_MAX: t.CDefine = 5
_dl_err_store: t.CArray[t.CChar, 512]
def last_error() -> str:
"""返回最近一次下载失败原因文本（无则空串）。"""
return _dl_err_store
def _dl_set_err(text: str):
if text is None:
return
_dl_err_store[0] = 0
stdio.snprintf(_dl_err_store, 512, "%s", text)
def _dl_strdup(s: str) -> str:
"""stdlib 堆上复制字符串（调用方 free），None 失败。"""
if s is None:
return None
n: t.CSizeT = string.strlen(s)
out: str = stdlib.malloc(n + 1)
if out is None:
return None
if n > 0:
string.memcpy(out, s, n)
out[n] = 0
return out
# ============================================================
# 单切片拉取（含重定向跟随，≤5 跳；http/https 按 scheme 分发）
# ============================================================
def _dl_slice(url: str, pos: t.CSizeT, want: INT, verify: bool, ca_file: str,
dst: str, dst_cap: INT, out_total: t.CSizeT | t.CPtr) -> INT:
"""拉取 [pos, pos+want) 一片，返回体字节数写入 dst（dst 内正文起始需
调用方用 _dl_body_off 定位）；-2 重定向超限；-3 网络/服务器失败。"""
if url is None or dst is None or out_total is None:
return -1
cur: str = _dl_strdup(url)
if cur is None:
return -1
hops: INT = 0
rc: INT = 0
while True:
is_https: INT = 1 if (string.strncmp(cur, "https://", 8) == 0) else 0
st: INT = 0
tot: t.CSizeT = 0
loc: t.CArray[t.CChar, 512]
if is_https != 0:
rc = rhttps._DlOnce(cur, pos, want, verify, ca_file, dst, dst_cap,
c.Addr(st), loc, 512, c.Addr(tot))
else:
rc = requests._DlOnce(cur, pos, want, dst, dst_cap,
c.Addr(st), loc, 512, c.Addr(tot))
if rc == -2 and loc[0] != 0:
hops += 1
if hops > DL_REDIRECT_MAX:
_dl_set_err("too many redirects")
stdlib.free(cur)
return -2
new_url: str = stdlib.malloc(2048)
if new_url is None:
stdlib.free(cur)
return -1
if requests._ResolveRedirectUrl(cur, loc, new_url, 2048) != 0:
stdlib.free(new_url)
stdlib.free(cur)
_dl_set_err("bad redirect location")
return -3
stdlib.free(cur)
cur = new_url
continue
break
stdlib.free(cur)
if rc < 0:
return rc
c.Set(c.Deref(out_total), tot)
return rc
def _dl_body_off(dst: str, dst_cap: INT) -> INT:
"""定位响应体（头结束处）在 dst 中的偏移，找不到返回 -1。"""
return requests._FindBodyStart(dst, dst_cap)
# ============================================================
# 文件写入（按偏移落盘，支持多线程并发写不同区段）
# ============================================================
def _dl_write_at(path: str, src: str, length: INT, offset: t.CSizeT) -> INT:
"""打开文件（不存在则创建），Seek 到 offset 写 length 字节。0 成功；-1 失败。"""
if path is None or src is None or length <= 0:
return -1
h: wb.HANDLE = wf.CreateFileA(
path, wf.GENERIC_WRITE,
wf.FILE_SHARE_READ | wf.FILE_SHARE_WRITE | wf.FILE_SHARE_DELETE,
None, wf.OPEN_ALWAYS, wf.FILE_ATTRIBUTE_NORMAL, None)
if h == wb.INVALID_HANDLE_VALUE:
return -1
li: wb.LARGE_INTEGER = wb.LARGE_INTEGER()
li.QuadPart = t.CLongLong(offset)
ok_seek: BOOL = wf.SetFilePointerEx(h, c.Addr(li), None, wf.FILE_BEGIN)
if ok_seek == 0:
wb.CloseHandle(h)
return -1
written: ULONG = 0
ok_w: BOOL = wf.WriteFile(h, src, t.CUnsignedLong(length), c.Addr(written), None)
wb.CloseHandle(h)
if ok_w == 0:
return -1
if t.CInt(written) != length:
return -1
return 0
def _dl_truncate(path: str) -> INT:
"""CREATE_ALWAYS 建（截断）空文件。0 成功；-1 失败。"""
h: wb.HANDLE = wf.CreateFileA(
path, wf.GENERIC_WRITE,
wf.FILE_SHARE_READ | wf.FILE_SHARE_WRITE | wf.FILE_SHARE_DELETE,
None, wf.CREATE_ALWAYS, wf.FILE_ATTRIBUTE_NORMAL, None)
if h == wb.INVALID_HANDLE_VALUE:
return -1
wb.CloseHandle(h)
return 0
# ============================================================
# 探测：总大小 / Range 支持（首片 bytes=0-0）
# ============================================================
def _dl_probe(url: str, verify: bool, ca_file: str,
out_total: t.CSizeT | t.CPtr) -> INT:
"""探测服务器总大小与 Range 支持。
返回 0=支持 Range 且 out_total 有效；1=服务器返回 200（不支持 Range，
退化为整段下载，out_total=Content-Length）；-3 失败。"""
c.Set(c.Deref(out_total), 0)
buf: str = stdlib.malloc(8192)
if buf is None:
_dl_set_err("out of memory")
return -3
tot: t.CSizeT = 0
rc: INT = _dl_slice(url, 0, 0, verify, ca_file, buf, 8192, c.Addr(tot))
stdlib.free(buf)
if rc < 0:
_dl_set_err("probe failed")
return -3
if tot > 0:
c.Set(c.Deref(out_total), tot)
return 0
_dl_set_err("cannot determine size")
return -3
# ============================================================
# 下载一个区间带（[b_start, b_end)），逐片拉取并落盘
# ============================================================
def _dl_download_band(path: str, url: str, is_https: INT, verify: bool, ca_file: str,
b_start: t.CSizeT, b_end: t.CSizeT, chunk: INT,
buf: str, buf_cap: INT,
p_done: t.CSizeT | t.CPtr) -> INT:
"""0 成功；-1 网络/服务器；-2 写入失败。p_done 累计已写字节。"""
pos: t.CSizeT = b_start
while pos < b_end:
want: INT = chunk
if b_end - pos < t.CSizeT(want):
want = INT(b_end - pos)
tot: t.CSizeT = 0
rc: INT = _dl_slice(url, pos, want, verify, ca_file, buf, buf_cap, c.Addr(tot))
if rc <= 0:
_dl_set_err("slice fetch failed")
return -1
bo: INT = _dl_body_off(buf, buf_cap)
if bo < 0:
_dl_set_err("bad slice response")
return -1
got: INT = rc
if bo + got > buf_cap:
got = buf_cap - bo
if got <= 0:
_dl_set_err("slice empty")
return -1
if _dl_write_at(path, buf + bo, got, pos) != 0:
_dl_set_err("write file failed")
return -2
c.Set(c.Deref(p_done), c.Deref(p_done) + t.CSizeT(got))
pos += t.CSizeT(got)
if got < want:
# 服务端提前结束（正常收尾时仅末片 < want）
if pos >= b_end:
break
_dl_set_err("short slice")
return -1
return 0
# ============================================================
# 多线程 worker
# ============================================================
@t.Object
class DlJob:
path: str
url: str
is_https: INT
verify: INT
ca_file: str
start: t.CSizeT
end: t.CSizeT
chunk: INT
buf: str
buf_cap: INT
done: t.CSizeT
status: INT
def _dl_worker(arg: VOIDPTR) -> t.CPtr:
if arg is None:
return None
job: DlJob | t.CPtr = arg
rc: INT = _dl_download_band(
job.path, job.url, job.is_https,
True if job.verify != 0 else False,
job.ca_file,
job.start, job.end, job.chunk, job.buf, job.buf_cap,
c.Addr(job.done))
if rc != 0:
job.status = rc
return None
# ============================================================
# 公共 API
# ============================================================
def download(url: str, path: str, threads: INT = 1,
chunk_size: INT = DL_DEFAULT_CHUNK,
verify: bool = True, ca_file: str = None) -> INT:
"""大文件切片下载。
Args:
url: 支持 http:// 与 https://（Range 分段拉取）
path: 保存路径（不存在自动创建；已存在则截断重建）
threads: 1=串行切片；2~8=多线程并行分带下载
chunk_size: 单片大小（字节，默认 256KB）
verify / ca_file: https 证书校验参数（对齐 requests.https）
Returns:
0 成功；-1 参数错误；-2 文件创建失败；-3 服务器不支持/下载失败；
-4 写入失败。last_error() 可查原因文本。
"""
_dl_err_store[0] = 0
if url is None or path is None or url[0] == 0 or path[0] == 0:
_dl_set_err("url/path empty")
return -1
if chunk_size < 1024:
chunk_size = DL_DEFAULT_CHUNK
if threads < 1:
threads = 1
if threads > DL_MAX_THREADS:
threads = DL_MAX_THREADS
if string.strncmp(url, "http://", 7) != 0 \
and string.strncmp(url, "https://", 8) != 0:
_dl_set_err("unsupported scheme")
return -1
# 探测总大小与 Range 支持
total: t.CSizeT = 0
prc: INT = _dl_probe(url, verify, ca_file, c.Addr(total))
if prc < 0:
return -3
if total == 0:
_dl_set_err("cannot determine size")
return -3
# 建立（截断）目标文件
if _dl_truncate(path) != 0:
_dl_set_err("create file failed")
return -2
stdio.printf("[download] %s -> %s total=%zu bytes threads=%d chunk=%d\n",
url, path, total, threads, chunk_size)
slab: INT = chunk_size + 8192
if threads == 1:
# ---- 串行切片 ----
buf: str = stdlib.malloc(t.CSizeT(slab))
if buf is None:
_dl_set_err("out of memory")
return -3
done: t.CSizeT = 0
rc: INT = _dl_download_band(path, url, 0, verify, ca_file,
0, total, chunk_size, buf, slab, c.Addr(done))
stdlib.free(buf)
if rc != 0:
return -3 if rc == -1 else -4
stdio.printf("[download] done: %zu/%zu bytes\n", done, total)
return 0
# ---- 多线程并行分带 ----
band: t.CSizeT = (total + t.CSizeT(threads) - 1) / t.CSizeT(threads)
ctxs: t.CArray[VOIDPTR, DL_MAX_THREADS]
handles: t.CArray[VOIDPTR, DL_MAX_THREADS]
threads_ok: INT = 0
i: INT = 0
while i < threads:
b_start: t.CSizeT = t.CSizeT(i) * band
if b_start >= total:
break
b_end: t.CSizeT = b_start + band
if b_end > total:
b_end = total
job: DlJob | t.CPtr = stdlib.calloc(1, DlJob.__sizeof__())
if job is None:
break
job.path = _dl_strdup(path)
job.url = _dl_strdup(url)
if job.path is None or job.url is None:
stdlib.free(job.path)
stdlib.free(job.url)
stdlib.free(t.CVoid(t.CUInt64T(job), t.CPtr))
break
job.is_https = 1 if (string.strncmp(url, "https://", 8) == 0) else 0
job.verify = 1 if verify else 0
job.ca_file = _dl_strdup(ca_file) if ca_file is not None else None
job.start = b_start
job.end = b_end
job.chunk = chunk_size
job.buf_cap = slab
job.buf = stdlib.malloc(t.CSizeT(slab))
if job.buf is None:
stdlib.free(job.path)
stdlib.free(job.url)
if job.ca_file is not None:
stdlib.free(job.ca_file)
stdlib.free(t.CVoid(t.CUInt64T(job), t.CPtr))
break
job.done = 0
job.status = 0
ctxs[i] = t.CVoid(t.CUInt64T(job), t.CPtr)
th: vt.Thread | t.CPtr = vt.Thread(_dl_worker, ctxs[i])
if th is None:
stdlib.free(job.buf)
stdlib.free(job.path)
stdlib.free(job.url)
if job.ca_file is not None:
stdlib.free(job.ca_file)
stdlib.free(t.CVoid(t.CUInt64T(job), t.CPtr))
break
th.start()
handles[i] = th.handle
threads_ok += 1
i += 1
if threads_ok == 0:
_dl_set_err("thread create failed")
return -3
# 等待全部线程结束
j: INT = 0
while j < threads_ok:
wsync.WaitForSingleObject(handles[j], 0xFFFFFFFF)
wb.CloseHandle(handles[j])
j += 1
# 汇总结果
total_done: t.CSizeT = 0
failed_job: INT = 0
k: INT = 0
while k < threads_ok:
jobk: DlJob | t.CPtr = ctxs[k]
total_done += jobk.done
if jobk.status != 0:
failed_job = 1
k += 1
# 释放线程上下文
m: INT = 0
while m < threads_ok:
jobm: DlJob | t.CPtr = ctxs[m]
stdlib.free(jobm.path)
stdlib.free(jobm.url)
if jobm.ca_file is not None:
stdlib.free(jobm.ca_file)
stdlib.free(jobm.buf)
stdlib.free(t.CVoid(t.CUInt64T(jobm), t.CPtr))
m += 1
if failed_job != 0 or total_done != total:
stdio.printf("[download] incomplete: %zu/%zu bytes\n", total_done, total)
_dl_set_err("download incomplete")
return -3
stdio.printf("[download] done: %zu/%zu bytes\n", total_done, total)
return 0
requests/https.vp
+664
View File
@@ -0,0 +1,664 @@
import t, c
from stdint import *
import string
import socket
import stdio
import stdlib
import requests
import openssl
# ============================================================
# requests/https.vp — HTTPS 客户端（openssl SSL/TLS + 证书核验）
#
# 与 requests（HTTP）同名 API，但走 TLS 并默认校验对端证书：
# import requests.https as rhttps
# resp = rhttps.get("https://example.com/", verify=True)
# if resp is not None:
# print(resp.status_code)
# print(resp.text)
# requests.release(resp)
#
# 说明:
# - verify=True 时校验证书链 + 主机名（hostname 校验 SSL_set1_host），
# 提供 ca_file 指向 PEM CA bundle；不提供则走系统默认路径
# （Windows 上通常为空，真实站点会因"无法验证证书链"失败返回 None）。
# - 失败时返回 None，可用 last_error() 获取原因文本。
# - 握手成功后可用 last_tls_info() 查看 TLS 版本 / 加密套件 / 对端证书主题。
# ============================================================
HTTPS_PORT: t.CDefine = 443
# 模块内部状态
_ossl_ready: bool = False
_socket_ready: bool = False
_err_store: t.CArray[t.CChar, 512] # 最近一次失败原因（静态缓冲）
_info_store: t.CArray[t.CChar, 512] # 最近一次成功握手的 TLS 元信息
_scratch_store: t.CArray[t.CChar, 512] # 临时拼接缓冲
_last_verify_val: INT = -1 # 最近一次握手后 SSL_get_verify_result（0=通过；-1=未启用）
_last_verify_flag: INT = -1 # 最近一次请求 verify 标志（1=开启；0=关闭；-1=未知）
def last_verify_result() -> int:
"""返回最近一次握手的证书校验结果：0 通过；>0 失败码；-1 未开启校验。"""
return _last_verify_val
def last_verify_enabled() -> int:
"""返回最近一次请求是否开启校验（1=是；0=否；-1=未知）。"""
return _last_verify_flag
# ============================================================
# 内部辅助
# ============================================================
def last_error() -> str:
"""返回最近一次失败的原因文本（无则空串）。"""
return _err_store
def last_tls_info() -> str:
"""返回最近一次成功握手的 TLS 信息（版本/套件/对端主题），无则空串。"""
return _info_store
def _RecordErrorText(text: str):
"""把失败原因写入静态缓冲 _err_store。"""
if text is None: return
_err_store[0] = 0
stdio.snprintf(_err_store, 512, "%s", text)
def _RecordInfoText(text: str):
"""把 TLS 元信息写入静态缓冲 _info_store。"""
if text is None: return
_info_store[0] = 0
stdio.snprintf(_info_store, 512, "%s", text)
def _EnsureReady() -> INT:
"""初始化 socket 与 openssl（均幂等）。"""
global _socket_ready, _ossl_ready
if not _socket_ready:
if socket.SocketInit() != 0:
_RecordErrorText("socket init failed")
return -1
_socket_ready = True
if not _ossl_ready:
if openssl.OPENSSL_init_ssl(0, None) != 1:
_RecordErrorText("OPENSSL_init_ssl failed")
return -1
_ossl_ready = True
return 0
def _ParseHttpsUrl(url: str, host_buf: str, host_size: INT, path_buf: str, path_size: INT, port: INT | t.CPtr) -> INT:
"""解析 https://host[:port][/path]，端口缺省 443。"""
if url is None: return -1
if host_buf is None or path_buf is None or port is None: return -1
p: str = url
# Skip "https://"（8 字节）
if string.strncmp(p, "https://", 8) == 0:
p += 8
# Parse host
host_start: str = p
host_len: INT = 0
while p[0] != '\0' and p[0] != ':' and p[0] != '/':
host_len += 1
p += 1
if host_len == 0 or host_len >= host_size: return -1
string.strncpy(host_buf, host_start, host_len)
host_buf[host_len] = 0
# Parse port
if p[0] == ':':
p += 1
port_val: INT = 0
while p[0] >= '0' and p[0] <= '9':
port_val = port_val * 10 + INT(p[0] - '0')
p += 1
c.DerefAs(port, port_val)
else:
c.DerefAs(port, HTTPS_PORT)
# Parse path
if p[0] == '/':
path_start: str = p
path_len: INT = 0
while p[0] != '\0':
path_len += 1
p += 1
if path_len >= path_size: return -1
string.strncpy(path_buf, path_start, path_len)
path_buf[path_len] = 0
else:
if path_size < 2: return -1
path_buf[0] = '/'
path_buf[1] = 0
return 0
def _ScratchBuf() -> str:
"""返回静态拼接缓冲（512B，复用）。"""
return _scratch_store
def _TlsErrText():
"""把最近一次 openssl 错误队列文本写入静态 scratch 缓冲。"""
e: t.CUInt32T = openssl.ERR_get_error()
if e == 0:
_scratch_store[0] = 0
string.strncpy(_scratch_store, "openssl error (no detail)", 26)
return
openssl.ERR_error_string_n(e, _scratch_store, 512)
# ============================================================
# 核心请求
# ============================================================
def _HttpsRequest(method: str, url: str, verify: bool, ca_file: str, data: str, content_type: str) -> requests.Response | t.CPtr:
"""HTTPS GET/POST 核心实现。
verify=False 时跳过证书链/主机名校验（不推荐）；verify=True 时
ca_file 提供 PEM CA bundle，否则使用系统默认 CA 路径。
"""
global _last_verify_val, _last_verify_flag
if requests._mbuddy is None: return None
_last_verify_flag = 1 if verify else 0
resp: requests.Response | t.CPtr = requests.create_response()
if resp is None: return None
if _EnsureReady() != 0:
requests.release(resp)
return None
# 临时缓冲（错误文本格式化；openssl 详情走模块级 scratch）
errmsg: str = requests._mbuddy.alloc(512)
if errmsg is None:
requests.release(resp)
return None
# ---- 解析 URL ----
host_buf: t.CArray[t.CChar, 256]
path_buf: t.CArray[t.CChar, 1024]
port: INT = 0
if _ParseHttpsUrl(url, host_buf, 256, path_buf, 1024, c.Addr(port)) != 0:
stdio.snprintf(errmsg, 512, "bad https url: %s", url)
_RecordErrorText(errmsg)
requests._mbuddy.free(errmsg)
requests.release(resp)
return None
# ---- 建立 TCP 连接 ----
sock = socket.Socket(socket.AF_INET, socket.SOCK_STREAM)
if sock.fd == socket.INVALID_SOCKET:
_RecordErrorText("socket() failed")
requests._mbuddy.free(errmsg)
requests.release(resp)
return None
if sock.connect(host_buf, port) != 0:
stdio.snprintf(errmsg, 512, "connect %s:%d failed", host_buf, port)
_RecordErrorText(errmsg)
sock.close()
requests._mbuddy.free(errmsg)
requests.release(resp)
return None
# ---- SSL_CTX / SSL ----
ctx: openssl.SslCtxPtr = openssl.SSL_CTX_new(openssl.TLS_client_method())
if ctx is None:
_RecordErrorText("SSL_CTX_new failed")
sock.close()
requests._mbuddy.free(errmsg)
requests.release(resp)
return None
if verify:
openssl.SSL_CTX_set_verify(ctx, openssl.SslVerifyPeer, None)
if ca_file is not None and ca_file[0] != 0:
if openssl.SSL_CTX_load_verify_locations(ctx, ca_file, None) != 1:
_TlsErrText()
stdio.snprintf(errmsg, 512, "load CA file failed: %s (%s)", ca_file, _ScratchBuf())
_RecordErrorText(errmsg)
openssl.SSL_CTX_free(ctx)
sock.close()
requests._mbuddy.free(errmsg)
requests.release(resp)
return None
else:
# 系统默认路径（Windows 上通常为空，证书链校验将失败）
openssl.SSL_CTX_set_default_verify_paths(ctx)
else:
openssl.SSL_CTX_set_verify(ctx, openssl.SslVerifyNone, None)
ssl: openssl.SslPtr = openssl.SSL_new(ctx)
if ssl is None:
_RecordErrorText("SSL_new failed")
openssl.SSL_CTX_free(ctx)
sock.close()
requests._mbuddy.free(errmsg)
requests.release(resp)
return None
openssl.SSL_set_fd(ssl, sock.fd)
# SNI（SSL_set_tlsext_host_name 是宏 → 直接 SSL_ctrl）
openssl.SSL_ctrl(ssl, openssl.SslCtrlSetTlsextHostname, openssl.TlsextNametypeHostName, host_buf)
# 主机名校验（TLS 内核对证书 SAN/CN 与 host 比对）
if verify:
openssl.SSL_set1_host(ssl, host_buf)
# ---- 握手 ----
hret: INT = openssl.SSL_connect(ssl)
if hret <= 0:
want: INT = openssl.SSL_get_error(ssl, hret)
_TlsErrText()
stdio.snprintf(errmsg, 512, "SSL_connect failed (want=%d): %s", want, _ScratchBuf())
_RecordErrorText(errmsg)
openssl.SSL_free(ssl)
openssl.SSL_CTX_free(ctx)
sock.close()
requests._mbuddy.free(errmsg)
requests.release(resp)
return None
# ---- 记录最近一次证书校验结果（供 last_verify_result 查询）----
if verify:
_last_verify_val = INT(openssl.SSL_get_verify_result(ssl))
else:
_last_verify_val = -1
# ---- 证书链校验结果复核 ----
if verify:
vres: t.CInt32T = openssl.SSL_get_verify_result(ssl)
if vres != 0:
reason: str = openssl.X509_verify_cert_error_string(vres)
if reason is None:
stdio.snprintf(errmsg, 512, "SSL certificate verify failed: error %d", INT(vres))
else:
stdio.snprintf(errmsg, 512, "SSL certificate verify failed (%d): %s", INT(vres), reason)
_RecordErrorText(errmsg)
openssl.SSL_free(ssl)
openssl.SSL_CTX_free(ctx)
sock.close()
requests._mbuddy.free(errmsg)
requests.release(resp)
return None
# ---- 记录 TLS 元信息（版本 / 套件 / 对端证书主题）----
name_buf: str = requests._mbuddy.alloc(512)
if name_buf is not None:
peer: openssl.X509Ptr = openssl.SSL_get1_peer_certificate(ssl)
if peer is not None:
subj: openssl.X509NamePtr = openssl.X509_get_subject_name(peer)
if subj is not None:
name_buf[0] = 0
openssl.X509_NAME_oneline(subj, name_buf, 512)
name_buf[511] = 0
openssl.X509_free(peer)
else:
name_buf[0] = 0
string.strncpy(name_buf, "(no peer certificate)", 21)
name_buf[21] = 0
ver: str = openssl.SSL_get_version(ssl)
cipher: str = openssl.SSL_CIPHER_get_name(openssl.SSL_get_current_cipher(ssl))
stdio.snprintf(errmsg, 512, "%s | cipher %s | peer: %s",
ver if ver is not None else "?", cipher if cipher is not None else "?", name_buf)
_RecordInfoText(errmsg)
requests._mbuddy.free(name_buf)
# ---- 组请求头并发送 ----
req_buf: str = requests._mbuddy.alloc(requests.HTTP_BUFSIZ)
if req_buf is None:
_RecordErrorText("alloc request buffer failed")
openssl.SSL_free(ssl)
openssl.SSL_CTX_free(ctx)
sock.close()
requests._mbuddy.free(errmsg)
requests.release(resp)
return None
body_len: INT = 0
is_post: bool = string.strcmp(method, "POST") == 0
if is_post:
body_len = INT(string.strlen(data)) if data is not None else 0
ct: str = content_type if content_type is not None else "application/x-www-form-urlencoded"
stdio.snprintf(req_buf, requests.HTTP_BUFSIZ,
"POST %s HTTP/1.1\r\nHost: %s\r\nContent-Type: %s\r\nContent-Length: %d\r\nAccept-Encoding: gzip\r\nConnection: close\r\n\r\n",
path_buf, host_buf, ct, body_len)
else:
stdio.snprintf(req_buf, requests.HTTP_BUFSIZ,
"GET %s HTTP/1.1\r\nHost: %s\r\nAccept-Encoding: gzip\r\nConnection: close\r\n\r\n",
path_buf, host_buf)
hdr_len: INT = INT(string.strlen(req_buf))
sent: INT = openssl.SSL_write(ssl, req_buf, hdr_len)
if sent <= 0:
_TlsErrText()
stdio.snprintf(errmsg, 512, "SSL_write header failed: %s", _ScratchBuf())
_RecordErrorText(errmsg)
requests._mbuddy.free(req_buf)
openssl.SSL_free(ssl)
openssl.SSL_CTX_free(ctx)
sock.close()
requests._mbuddy.free(errmsg)
requests.release(resp)
return None
if is_post and body_len > 0 and data is not None:
openssl.SSL_write(ssl, data, body_len)
requests._mbuddy.free(req_buf)
# ---- 接收响应 ----
resp_buf: str = requests._mbuddy.alloc(requests.HTTP_BUFSIZ)
if resp_buf is None:
_RecordErrorText("alloc response buffer failed")
openssl.SSL_free(ssl)
openssl.SSL_CTX_free(ctx)
sock.close()
requests._mbuddy.free(errmsg)
requests.release(resp)
return None
total: INT = 0
while total < requests.HTTP_BUFSIZ - 1:
n: INT = openssl.SSL_read(ssl, resp_buf + total, requests.HTTP_BUFSIZ - 1 - total)
if n <= 0: break
total += n
resp_buf[total] = 0
# 关闭 TLS 与 TCP
openssl.SSL_shutdown(ssl)
openssl.SSL_free(ssl)
openssl.SSL_CTX_free(ctx)
sock.close()
resp._raw_buf = resp_buf
requests.parse_response(resp, total)
requests._mbuddy.free(errmsg)
return resp
# ============================================================
# 公共 API
# ============================================================
def _HttpsRedirectLoop(method: str, url: str, verify: bool, ca_file: str, data: str, content_type: str) -> requests.Response | t.CPtr:
"""HTTPS GET/POST + 重定向跟随（跨 http/https，最多 5 跳）。"""
if requests._mbuddy is None: return None
cur: str = requests._mbuddy.alloc(2048)
if cur is None: return None
if INT(string.strlen(url)) >= 2047:
_RecordErrorText("url too long")
requests._mbuddy.free(cur)
return None
string.strcpy(cur, url)
hop: INT = 0
while hop < 5:
resp: requests.Response | t.CPtr = _HttpsRequest(method, cur, verify, ca_file, data, content_type)
if resp is None:
requests._mbuddy.free(cur)
return None
code: INT = resp.status_code
if not (code >= 300 and code < 400):
requests._mbuddy.free(cur)
return resp
loc: str = requests._mbuddy.alloc(2048)
if loc is None:
requests._mbuddy.free(cur)
return resp
if requests.header_value(resp, "Location", loc, 2048) != 0:
requests._mbuddy.free(loc)
requests._mbuddy.free(cur)
return resp
nxt: str = requests._mbuddy.alloc(2048)
if nxt is None:
requests._mbuddy.free(loc)
requests._mbuddy.free(cur)
return resp
if requests._ResolveRedirectUrl(cur, loc, nxt, 2048) != 0:
_RecordErrorText("unresolvable redirect Location")
requests.release(resp)
requests._mbuddy.free(nxt)
requests._mbuddy.free(loc)
requests._mbuddy.free(cur)
return None
requests.release(resp)
# 跨协议分流：https → 继续本循环；http → 交给 requests（http）链
if string.strncmp(nxt, "https://", 8) != 0:
requests._mbuddy.free(loc)
requests._mbuddy.free(cur)
http_resp: requests.Response | t.CPtr = requests.get(nxt)
requests._mbuddy.free(nxt)
return http_resp
requests._mbuddy.free(loc)
requests._mbuddy.free(cur)
cur = nxt
hop += 1
_RecordErrorText("too many redirects")
requests._mbuddy.free(cur)
return None
def get(url: str, verify: bool = True, ca_file: str = None) -> requests.Response | t.CPtr:
"""HTTPS GET request（自动跟随 3xx 重定向，最多 5 跳）。
Args:
url: Full URL (e.g., "https://example.com/path")
verify: True 时校验证书链 + 主机名（默认）
ca_file: PEM CA bundle 路径；None 时使用系统默认 CA 路径
Returns:
Pointer to a Response object, None on failure (see last_error()).
"""
return _HttpsRedirectLoop("GET", url, verify, ca_file, None, None)
def post(url: str, data: str = None, content_type: str = "application/x-www-form-urlencoded",
verify: bool = True, ca_file: str = None) -> requests.Response | t.CPtr:
"""HTTPS POST request（自动跟随 3xx 重定向，最多 5 跳）。
Args:
url: Full URL
data: Request body data
content_type: Content-Type header
verify: True 时校验证书链 + 主机名（默认）
ca_file: PEM CA bundle 路径；None 时使用系统默认 CA 路径
Returns:
Pointer to a Response object, None on failure (see last_error()).
"""
return _HttpsRedirectLoop("POST", url, verify, ca_file, data, content_type)
# ============================================================
# 大文件 Range 下载原语（requests.download 使用，与 requests._DlOnce 镜像）
#
# 线程安全：本函数不使用 _mbuddy（请求/响应缓冲一律 stdlib），每个连接独立
# SSL_CTX/SSL/socket，供多线程并行切片并发调用（OPENSSL_init_ssl 由主线程
# 先 _EnsureReady 预热一次）。
#
# 返回: >=0 响应体字节数；-1 参数错；-2 3xx 重定向（Location 写 out_loc）；
# -3 其它失败（last_error() 查看原因）。
# ============================================================
def _DlOnce(url: str, start: t.CSizeT, count: INT, verify: bool, ca_file: str,
dst: str, dst_cap: INT, out_status: INT | t.CPtr,
out_loc: str, out_loccap: INT,
out_total: t.CSizeT | t.CPtr) -> INT:
"""HTTPS 单次 Range 拉取（线程安全，见模块注释）。"""
global _last_verify_val, _last_verify_flag
if url is None or dst is None or dst_cap <= 0 \
or out_status is None or out_total is None:
return -1
c.Set(c.Deref(out_status), 0)
c.Set(c.Deref(out_total), 0)
if out_loc is not None and out_loccap > 0:
out_loc[0] = 0
_last_verify_flag = 1 if verify else 0
if _EnsureReady() != 0:
return -3
host_buf: t.CArray[t.CChar, 256]
path_buf: t.CArray[t.CChar, 1024]
port: INT = 0
if _ParseHttpsUrl(url, host_buf, 256, path_buf, 1024, c.Addr(port)) != 0:
_RecordErrorText("bad https url")
return -3
sock = socket.Socket(socket.AF_INET, socket.SOCK_STREAM)
if sock.fd == socket.INVALID_SOCKET:
_RecordErrorText("socket() failed")
return -3
if sock.connect(host_buf, port) != 0:
_RecordErrorText("connect failed")
sock.close()
return -3
ctx: openssl.SslCtxPtr = openssl.SSL_CTX_new(openssl.TLS_client_method())
if ctx is None:
_RecordErrorText("SSL_CTX_new failed")
sock.close()
return -3
if verify:
openssl.SSL_CTX_set_verify(ctx, openssl.SslVerifyPeer, None)
if ca_file is not None and ca_file[0] != 0:
if openssl.SSL_CTX_load_verify_locations(ctx, ca_file, None) != 1:
_RecordErrorText("load CA file failed")
openssl.SSL_CTX_free(ctx)
sock.close()
return -3
else:
openssl.SSL_CTX_set_default_verify_paths(ctx)
else:
openssl.SSL_CTX_set_verify(ctx, openssl.SslVerifyNone, None)
ssl: openssl.SslPtr = openssl.SSL_new(ctx)
if ssl is None:
_RecordErrorText("SSL_new failed")
openssl.SSL_CTX_free(ctx)
sock.close()
return -3
openssl.SSL_set_fd(ssl, sock.fd)
openssl.SSL_ctrl(ssl, openssl.SslCtrlSetTlsextHostname, openssl.TlsextNametypeHostName, host_buf)
if verify:
openssl.SSL_set1_host(ssl, host_buf)
hret: INT = openssl.SSL_connect(ssl)
if hret <= 0:
_RecordErrorText("SSL_connect failed")
openssl.SSL_free(ssl)
openssl.SSL_CTX_free(ctx)
sock.close()
return -3
if verify:
_last_verify_val = INT(openssl.SSL_get_verify_result(ssl))
if _last_verify_val != 0:
_RecordErrorText("SSL certificate verify failed")
openssl.SSL_free(ssl)
openssl.SSL_CTX_free(ctx)
sock.close()
return -3
else:
_last_verify_val = -1
# 组 Range 请求头（不带 Accept-Encoding，保证响应为原始字节）
req: str = stdlib.malloc(1024)
if req is None:
openssl.SSL_free(ssl)
openssl.SSL_CTX_free(ctx)
sock.close()
return -3
cnt: INT = count
if cnt < 0:
cnt = 0
end: t.CSizeT = start + t.CSizeT(cnt) - 1
if cnt <= 0:
end = start
stdio.snprintf(req, 1024,
"GET %s HTTP/1.1\r\nHost: %s\r\nRange: bytes=%zu-%zu\r\nConnection: close\r\n\r\n",
path_buf, host_buf, start, end)
sent: INT = openssl.SSL_write(ssl, req, INT(string.strlen(req)))
stdlib.free(req)
if sent <= 0:
_RecordErrorText("SSL_write failed")
openssl.SSL_free(ssl)
openssl.SSL_CTX_free(ctx)
sock.close()
return -3
total: INT = 0
tgt: INT = -1
while total < dst_cap - 1:
room: INT = dst_cap - 1 - total
if room > 4096:
room = 4096
n: INT = openssl.SSL_read(ssl, dst + total, room)
if n <= 0:
break
total += n
if tgt < 0:
bo: INT = requests._FindBodyStart(dst, total)
if bo >= 0:
cl: INT = requests._CLenOf(dst, total)
if cl >= 0:
tgt = bo + cl
if tgt >= 0 and total >= tgt:
break
openssl.SSL_shutdown(ssl)
openssl.SSL_free(ssl)
openssl.SSL_CTX_free(ctx)
sock.close()
dst[total] = 0
st: INT = requests._StatusRaw(dst)
c.Set(c.Deref(out_status), st)
if st == 0:
_RecordErrorText("empty response")
return -3
if requests._IsHttpRedirect(st):
if out_loc is not None and out_loccap > 0:
if requests._RawHeaderValue(dst, total, "Location", out_loc, out_loccap) != 0:
out_loc[0] = 0
return -2
if st != 200 and st != 206:
_RecordErrorText("http status not 200/206")
return -3
if st == 206:
c.Set(c.Deref(out_total), requests._CRTotalOf(dst, total))
else:
cl2: INT = requests._CLenOf(dst, total)
if cl2 > 0:
c.Set(c.Deref(out_total), t.CSizeT(cl2))
bo: INT = requests._FindBodyStart(dst, total)
if bo < 0 or bo >= total:
return 0
blen: INT = total - bo
if blen > cnt and cnt > 0:
blen = cnt
return blen