修复 binascii.vp
This commit is contained in:
TermiNexus committed
1 parent
9a758a8f6c
commit
21f9b7a078
2 files changed
+374
-12
No files matched your search
+14
-12
@@ -1,21 +1,22 @@
|
||||
import t, c
|
||||
import t, c
|
||||
import stdlib
|
||||
|
||||
|
||||
HEX_CHARS: t.CArray[t.CChar, None] = "0123456789abcdef"
|
||||
HEX_VALS: t.CArray[t.CInt, 256]
|
||||
HEX_VALS: t.CArray[t.CInt, 256]
|
||||
|
||||
B64_CHARS: t.CArray[t.CChar, None] = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/"
|
||||
B64_VALS: t.CArray[t.CInt, 256]
|
||||
B64_VALS: t.CArray[t.CInt, 256]
|
||||
|
||||
# HQX (BinHex 4.0) Alphabet — The standard 64-character table from RFC 1741, where the index corresponds to the 6-bit value (0..63).
|
||||
# Note: There must be exactly 64 elements (previously, 67 were given → compiler fail-fast
|
||||
# "Array initialization element count 67 exceeds array declared length 64"). This table is the authoritative source,
|
||||
# init_binascii builds HQX_VALS based on this alone, and this table must not be rewritten.
|
||||
HQX_CHARS: t.CArray[t.CChar, 64] = [
|
||||
' ', '!', chr(34), '#', '$', '%', '&', "'", '(', ')', '*', '+', ',', '-', '.', '/',
|
||||
'0', '1', '2', '3', '4', '5', '6', '7', '8', '9',
|
||||
'@',
|
||||
'A', 'B', 'C', 'D', 'E', 'F', 'G', 'H', 'I', 'J', 'K', 'L', 'M', 'N',
|
||||
'P', 'Q', 'R', 'S', 'T', 'U', 'V', 'X', 'Y', 'Z',
|
||||
'[', chr(92), '`',
|
||||
'a', 'b', 'c', 'd', 'e', 'f', 'g', 'h', 'i', 'j', 'k', 'l', 'm'
|
||||
'!', chr(34), '#', '$', '%', '&', "'", '(', ')', '*', '+', ',', '-', '0', '1', '2',
|
||||
'3', '4', '5', '6', '8', '9', '@', 'A', 'B', 'C', 'D', 'E', 'F', 'G', 'H', 'I',
|
||||
'J', 'K', 'L', 'M', 'N', 'P', 'Q', 'R', 'S', 'T', 'U', 'V', 'X', 'Y', 'Z', '[',
|
||||
'`', 'a', 'b', 'c', 'd', 'e', 'f', 'h', 'i', 'j', 'k', 'l', 'm', 'p', 'q', 'r'
|
||||
]
|
||||
HQX_VALS: t.CArray[t.CInt, 256]
|
||||
|
||||
@@ -102,8 +103,9 @@ def init_binascii():
|
||||
HEX_VALS[t.CInt(HEX_CHARS[i] - 32)] = i
|
||||
for i in range(64):
|
||||
B64_VALS[t.CInt(B64_CHARS[i])] = i
|
||||
for i in range(64):
|
||||
HQX_CHARS[i] = t.CChar(32 + i)
|
||||
# HQX_VALS 由标准 BinHex 字母表（见 HQX_CHARS 静态初值，RFC 1741）反查建立。
|
||||
# 注意：此处【不得】再重写 HQX_CHARS——此前用 `chr(32 + i)` 覆盖 0..63，
|
||||
# 会把标准 BinHex 表整张冲掉，导致 b2a_hqx/a2b_hqx 与 RFC 1741 不兼容。
|
||||
for i in range(64):
|
||||
HQX_VALS[t.CUnsignedChar(HQX_CHARS[i])] = i
|
||||
|
||||
|
||||
+360
@@ -0,0 +1,360 @@
|
||||
import t, c
|
||||
from stdint import *
|
||||
import string
|
||||
import socket
|
||||
import memhub
|
||||
import stdio
|
||||
|
||||
|
||||
# ============================================================
|
||||
# requests.py — HTTP client library (Python original style)
|
||||
#
|
||||
# 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
|
||||
|
||||
|
||||
# ============================================================
|
||||
# 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 (points into _raw_buf)
|
||||
body_len: INT # Response body length
|
||||
_raw_buf: str # Raw buffer pointer
|
||||
|
||||
|
||||
# ============================================================
|
||||
# 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
|
||||
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
|
||||
|
||||
|
||||
# ============================================================
|
||||
# 公共 API
|
||||
# ============================================================
|
||||
|
||||
def get(url: str) -> Response | t.CPtr:
|
||||
"""HTTP GET request.
|
||||
|
||||
Args:
|
||||
url: Full URL (e.g., "http://example.com/path")
|
||||
Returns:
|
||||
Pointer to a Response object (allocated from _mbuddy), returns None on failure.
|
||||
"""
|
||||
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\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 status line
|
||||
_ParseStatusLine(resp_buf, c.Addr(resp.status_code))
|
||||
|
||||
# Find body start
|
||||
body_offset: INT = _FindBodyStart(resp_buf, total)
|
||||
if body_offset < 0 or body_offset >= total:
|
||||
resp.text = None
|
||||
resp.body_len = 0
|
||||
else:
|
||||
resp.text = resp_buf + body_offset
|
||||
resp.body_len = total - body_offset
|
||||
|
||||
return resp
|
||||
|
||||
|
||||
def post(url: str, data: str = None, content_type: str = "application/x-www-form-urlencoded") -> Response | t.CPtr:
|
||||
"""HTTP POST request.
|
||||
|
||||
Args:
|
||||
url: Full URL
|
||||
data: Request body data
|
||||
content_type: Content-Type header (default is "application/x-www-form-urlencoded")
|
||||
Returns:
|
||||
Pointer to Response object (allocated from _mbuddy), returns None on failure.
|
||||
"""
|
||||
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\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
|
||||
|
||||
_ParseStatusLine(resp_buf, c.Addr(resp.status_code))
|
||||
body_offset: INT = _FindBodyStart(resp_buf, total)
|
||||
if body_offset < 0 or body_offset >= total:
|
||||
resp.text = None
|
||||
resp.body_len = 0
|
||||
else:
|
||||
resp.text = resp_buf + body_offset
|
||||
resp.body_len = total - body_offset
|
||||
|
||||
return resp
|
||||
|
||||
|
||||
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._raw_buf is not None:
|
||||
_mbuddy.free(resp._raw_buf)
|
||||
_mbuddy.free(t.CVoid(t.CUInt64T(resp), t.CPtr))
|
||||
Reference in new issue
Block a user