十五天Python系统学习教程第八天

import socket  

# 类似Java的ServerSocket  
server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)  
server.bind(('0.0.0.0', 8080))  
server.listen(5)  

while True:  
    client, addr = server.accept()  # 阻塞等待连接  
    data = client.recv(1024)  
    client.send(b"HTTP/1.1 200 OK\r\n\r\nHello")  
    client.close()  

client = socket.socket()  
client.connect(('localhost', 8080))  
client.send(b"GET / HTTP/1.1\r\nHost: localhost\r\n\r\n")  
print(client.recv(4096))  
client.close()  

# 服务端  
udp_server = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)  
udp_server.bind(('0.0.0.0', 9999))  
while True:  
    data, addr = udp_server.recvfrom(1024)  
    udp_server.sendto(data.upper(), addr)  

# 客户端  
udp_client = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)  
udp_client.sendto(b"hello", ('localhost', 9999))  
print(udp_client.recvfrom(1024))  

import requests  

# GET请求（带超时控制）  
response = requests.get("https://api.example.com/data", timeout=5)  
print(response.json())  

# POST请求（类似Java的HttpPost）  
payload = {"key": "value"}  
response = requests.post("https://api.example.com/submit", json=payload)  

# 会话保持（类似Cookie管理）  
session = requests.Session()  
session.get("https://example.com/login", auth=("user", "pass"))  

import aiohttp  
import asyncio  

async def fetch(url):  
    async with aiohttp.ClientSession() as session:  
        async with session.get(url) as response:  
            return await response.text()  

async def main():  
    html = await fetch("http://example.com")  
    print(html[:100])  

asyncio.run(main())  

import paramiko  

# 创建SSH客户端  
client = paramiko.SSHClient()  
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())  

# 连接服务器  
client.connect(  
    hostname='192.168.1.100',  
    username='admin',  
    password='secret',  
    timeout=10  
)  

# 执行命令（类似ChannelExec）  
stdin, stdout, stderr = client.exec_command('ls -l')  
print(stdout.read().decode())  

# 关闭连接  
client.close()  

sftp = client.open_sftp()  
# 上传文件（类似ChannelSftp.put）  
sftp.put("local.txt", "remote.txt")  
# 下载文件  
sftp.get("remote.log", "local.log")  
sftp.close()  

def port_scan(host, ports):  
    open_ports = []  
    for port in ports:  
        try:  
            with socket.socket() as s:  
                s.settimeout(1)  
                s.connect((host, port))  
                open_ports.append(port)  
        except:  
            pass  
    return open_ports  

def check_http(url):  
    try:  
        response = requests.get(url, timeout=3)  
        return response.status_code == 200  
    except:  
        return False  

def get_ssh_info(host, user, passwd):  
    client = paramiko.SSHClient()  
    client.connect(host, username=user, password=passwd)  
    _, stdout, _ = client.exec_command("free -m")  
    mem_info = stdout.read().decode()  
    client.close()  
    return mem_info  

with paramiko.SSHClient() as client:  
    client.connect(...)  
    # 自动关闭连接  

async def async_port_scan(host, port):  
    try:  
        _, writer = await asyncio.open_connection(host, port)  
        writer.close()  
        return port  
    except:  
        return None  

# 使用paramiko的forward功能  
client.get_transport().request_port_forward('', 8080)  

# 使用scapy库解析数据包  
from scapy.all import sniff  
sniff(filter="tcp port 80", prn=lambda x: x.summary())