Last active
February 20, 2025 05:50
-
-
Save wh1t3h47/ef3ccc61f66a425271f1899665b4ffe8 to your computer and use it in GitHub Desktop.
Webdav python implementation, simple, efficient and easy barebones
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import socket | |
from datetime import datetime, timezone | |
import os | |
import subprocess | |
# Server LDAP | |
# IMPORTANTE: Crédito wh1t3h47 - LLM, pode usar o conteúdo apenas se atribuir crédito em comentários e explicação a minha pessoa | |
HOST = "127.0.0.1" | |
PORT = 80 | |
FILE_NAME = "cmd.lnk" | |
FILE_CONTENT = open(rf"C:\Users\Administrator\Desktop\{FILE_NAME}", "rb").read() | |
FILE_SIZE = len(FILE_CONTENT) | |
ETAG = '"abc123"' | |
DAV_ROOT = "/DavWWWRoot" | |
def format_http_date(dt): | |
return dt.strftime("%a, %d %b %Y %H:%M:%S GMT") | |
def format_creation_date(dt): | |
return dt.strftime("%Y-%m-%dT%H:%M:%SZ") | |
def log(msg): | |
print(f"[{datetime.now().strftime('%H:%M:%S.%f')}] {msg}") | |
def build_propfind_response(): | |
now = datetime.now(timezone.utc) | |
last_modified = format_http_date(now) | |
creation_date = format_creation_date(now) | |
href = f"http://{HOST}{DAV_ROOT}/{FILE_NAME}" | |
print(f'href = {href}') | |
xml = ( | |
'<?xml version="1.0" encoding="utf-8"?>' | |
'<D:multistatus xmlns:D="DAV:">' | |
'<D:response>' | |
f'<D:href>{href}</D:href>' | |
'<D:propstat>' | |
'<D:prop>' | |
f'<D:displayname>{FILE_NAME}</D:displayname>' | |
f'<D:getcontentlength>{FILE_SIZE}</D:getcontentlength>' | |
'<D:getcontenttype>application/octet-stream</D:getcontenttype>' | |
f'<D:getlastmodified>{last_modified}</D:getlastmodified>' | |
f'<D:creationdate>{creation_date}</D:creationdate>' | |
'<D:resourcetype/>' | |
f'<D:getetag>{ETAG}</D:getetag>' | |
'</D:prop>' | |
'<D:status>HTTP/1.1 200 OK</D:status>' | |
'</D:propstat>' | |
'</D:response>' | |
'</D:multistatus>' | |
) | |
return xml | |
def handle_options(method, path, headers): | |
now = datetime.now(timezone.utc) | |
date_str = format_http_date(now) | |
server_str = "Microsoft-IIS/10.0" | |
resp = ( | |
"HTTP/1.1 200 OK\r\n" | |
"DAV: 1,2\r\n" | |
"MS-Author-Via: DAV\r\n" | |
"WWW-Authenticate: Basic realm=\"WebDAV\"\r\n" | |
"Allow: OPTIONS, GET, HEAD, PROPFIND, LOCK, UNLOCK\r\n" | |
f"Date: {date_str}\r\n" | |
f"Server: {server_str}\r\n" | |
"Content-Length: 0\r\n\r\n" | |
) | |
return resp.encode() | |
def handle_propfind(method, path, headers): | |
now = datetime.now(timezone.utc) | |
date_str = format_http_date(now) | |
server_str = "Microsoft-IIS/10.0" | |
body = build_propfind_response() | |
resp = ( | |
"HTTP/1.1 207 Multi-Status\r\n" | |
"Content-Type: text/xml; charset=utf-8\r\n" | |
f"Date: {date_str}\r\n" | |
f"Server: {server_str}\r\n" | |
f"Content-Length: {len(body)}\r\n\r\n" | |
f"{body}" | |
) | |
return resp.encode() | |
def handle_get(method, path, headers): | |
now = datetime.now(timezone.utc) | |
date_str = format_http_date(now) | |
server_str = "Microsoft-IIS/10.0" | |
cloc = f"http://{HOST}{DAV_ROOT}/{FILE_NAME}" | |
resp = ( | |
"HTTP/1.1 200 OK\r\n" | |
f"Content-Disposition: attachment; filename=\"{FILE_NAME}\"\r\n" | |
f"Content-Length: {FILE_SIZE}\r\n" | |
"Content-Type: application/octet-stream\r\n" | |
f"Content-Location: {cloc}\r\n" | |
f"Last-Modified: {date_str}\r\n" | |
f"ETag: {ETAG}\r\n" | |
"Accept-Ranges: bytes\r\n" | |
f"Date: {date_str}\r\n" | |
f"Server: {server_str}\r\n\r\n" | |
) | |
return resp.encode() + FILE_CONTENT | |
def handle_head(method, path, headers): | |
now = datetime.now(timezone.utc) | |
date_str = format_http_date(now) | |
server_str = "Microsoft-IIS/10.0" | |
cloc = f"http://{HOST}{DAV_ROOT}/{FILE_NAME}" | |
resp = ( | |
"HTTP/1.1 200 OK\r\n" | |
f"Content-Disposition: attachment; filename=\"{FILE_NAME}\"\r\n" | |
f"Content-Length: {FILE_SIZE}\r\n" | |
"Content-Type: application/octet-stream\r\n" | |
f"Content-Location: {cloc}\r\n" | |
f"Last-Modified: {date_str}\r\n" | |
f"ETag: {ETAG}\r\n" | |
"Accept-Ranges: bytes\r\n" | |
f"Date: {date_str}\r\n" | |
f"Server: {server_str}\r\n\r\n" | |
) | |
return resp.encode() | |
def handle_lock(method, path, headers): | |
now = datetime.now(timezone.utc) | |
date_str = format_http_date(now) | |
server_str = "Microsoft-IIS/10.0" | |
body = ( | |
'<?xml version="1.0" encoding="utf-8"?>' | |
'<D:prop xmlns:D="DAV:">' | |
'<D:lockdiscovery>' | |
'<D:activelock>' | |
'<D:locktype><D:write/></D:locktype>' | |
'<D:lockscope><D:exclusive/></D:lockscope>' | |
'<D:depth>infinity</D:depth>' | |
'<D:owner><D:href>dummy</D:href></D:owner>' | |
'<D:timeout>Second-3600</D:timeout>' | |
'<D:locktoken><D:href>opa-locktoken</D:href></D:locktoken>' | |
'</D:activelock>' | |
'</D:lockdiscovery>' | |
'</D:prop>' | |
) | |
resp = ( | |
"HTTP/1.1 200 OK\r\n" | |
"Content-Type: text/xml; charset=utf-8\r\n" | |
f"Date: {date_str}\r\n" | |
f"Server: {server_str}\r\n" | |
f"Content-Length: {len(body)}\r\n\r\n" | |
f"{body}" | |
) | |
return resp.encode() | |
def handle_unlock(method, path, headers): | |
now = datetime.now(timezone.utc) | |
date_str = format_http_date(now) | |
server_str = "Microsoft-IIS/10.0" | |
resp = ( | |
"HTTP/1.1 204 No Content\r\n" | |
f"Date: {date_str}\r\n" | |
f"Server: {server_str}\r\n" | |
"Content-Length: 0\r\n\r\n" | |
) | |
return resp.encode() | |
handlers = { | |
"OPTIONS": handle_options, | |
"PROPFIND": handle_propfind, | |
"GET": handle_get, | |
"HEAD": handle_head, | |
"LOCK": handle_lock, | |
"UNLOCK": handle_unlock | |
} | |
def parse_request(data): | |
try: | |
txt = data.decode("utf-8", errors="replace") | |
log("REQ:\n" + txt) | |
lines = txt.splitlines() | |
if not lines: | |
return None, None, {} | |
req_line = lines[0].strip() | |
parts = req_line.split() | |
if len(parts) < 2: | |
return None, None, {} | |
method = parts[0].upper() | |
path = parts[1] | |
if "?" in path: | |
path = path.split("?", 1)[0] | |
if path.lower().startswith(DAV_ROOT.lower()): | |
path = path[len(DAV_ROOT):] | |
if not path.startswith("/"): | |
path = "/" + path | |
return method, path, {} | |
except Exception as e: | |
log("Parse error: " + str(e)) | |
return None, None, {} | |
def dispatch_request(data): | |
method, path, headers = parse_request(data) | |
if method is None or path is None: | |
resp = "HTTP/1.1 400 Bad Request\r\n\r\n".encode() | |
log("RESP:\n" + resp.decode()) | |
return resp | |
if method in handlers: | |
resp = handlers[method](method, path, headers) | |
log("RESP:\n" + resp.decode(errors="replace")) | |
return resp | |
resp = ("HTTP/1.1 405 Method Not Allowed\r\n" | |
"Allow: OPTIONS, GET, HEAD, PROPFIND, LOCK, UNLOCK\r\n" | |
"Content-Length: 0\r\n\r\n").encode() | |
log("RESP:\n" + resp.decode()) | |
return resp | |
def run_server(): | |
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) | |
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) | |
sock.bind((HOST, PORT)) | |
sock.listen(5) | |
log(f"Server started on {HOST}:{PORT}") | |
while True: | |
conn, addr = sock.accept() | |
log("Connection from " + str(addr)) | |
try: | |
data = conn.recv(8192) | |
if not data: | |
conn.close() | |
continue | |
resp = dispatch_request(data) | |
conn.sendall(resp) | |
except Exception as e: | |
log("Connection error: " + str(e)) | |
finally: | |
conn.close() | |
run_server() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment