正文

网编程入门必备：30个实战案例，轻松学会网络编程技巧

/0 浏览量
0419

实战案例一：TCP客户端与服务器通信

在开始网络编程之前，了解TCP协议是基础。以下是一个简单的TCP客户端与服务器通信的例子：

# TCP服务器端代码
import socket

# 创建socket对象
server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

# 绑定端口
server_socket.bind(('127.0.0.1', 9999))

# 监听连接
server_socket.listen(5)

print("服务器启动，等待客户端连接...")

# 接受客户端连接
client_socket, client_address = server_socket.accept()

print(f"连接成功，客户端地址：{client_address}")

# 接收客户端数据
data = client_socket.recv(1024)
print(f"收到客户端数据：{data.decode()}")

# 发送数据给客户端
client_socket.sendall("Hello, client!")

# 关闭连接
client_socket.close()
server_socket.close()

# TCP客户端代码
import socket

# 创建socket对象
client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

# 连接服务器
client_socket.connect(('127.0.0.1', 9999))

# 发送数据给服务器
client_socket.sendall("Hello, server!")

# 接收服务器数据
data = client_socket.recv(1024)
print(f"收到服务器数据：{data.decode()}")

# 关闭连接
client_socket.close()

实战案例二：UDP客户端与服务器通信

UDP协议与TCP协议不同，它是一种无连接的协议。以下是一个简单的UDP客户端与服务器通信的例子：

# UDP服务器端代码
import socket

# 创建socket对象
server_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)

# 绑定端口
server_socket.bind(('127.0.0.1', 9999))

print("服务器启动，等待客户端连接...")

while True:
    # 接收客户端数据
    data, client_address = server_socket.recvfrom(1024)
    print(f"收到客户端数据：{data.decode()}")

    # 发送数据给客户端
    server_socket.sendto("Hello, client!".encode(), client_address)

# UDP客户端代码
import socket

# 创建socket对象
client_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)

# 连接服务器
client_socket.sendto("Hello, server!".encode(), ('127.0.0.1', 9999))

# 接收服务器数据
data = client_socket.recv(1024)
print(f"收到服务器数据：{data.decode()}")

# 关闭连接
client_socket.close()

实战案例三：HTTP服务器与客户端

HTTP协议是网络编程中常用的协议之一。以下是一个简单的HTTP服务器与客户端通信的例子：

# HTTP服务器端代码
import socket

# 创建socket对象
server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

# 绑定端口
server_socket.bind(('127.0.0.1', 8080))

# 监听连接
server_socket.listen(5)

print("HTTP服务器启动，等待客户端连接...")

while True:
    # 接受客户端连接
    client_socket, client_address = server_socket.accept()

    # 接收客户端请求
    request = client_socket.recv(1024).decode()
    print(f"收到客户端请求：{request}")

    # 构造响应数据
    response = "HTTP/1.1 200 OK\r\nContent-Type: text/html\r\n\r\nHello, world!"
    client_socket.sendall(response.encode())

    # 关闭连接
    client_socket.close()

# HTTP客户端代码
import socket

# 创建socket对象
client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

# 连接服务器
client_socket.connect(('127.0.0.1', 8080))

# 发送请求
client_socket.sendall("GET / HTTP/1.1\r\nHost: 127.0.0.1\r\n\r\n".encode())

# 接收响应
response = client_socket.recv(1024).decode()
print(f"收到服务器响应：{response}")

# 关闭连接
client_socket.close()

实战案例四：WebSocket通信

WebSocket协议是一种在单个TCP连接上进行全双工通信的协议。以下是一个简单的WebSocket客户端与服务器通信的例子：

# WebSocket服务器端代码
import asyncio
import websockets

async def echo(websocket, path):
    async for message in websocket:
        print(f"收到客户端消息：{message}")
        await websocket.send(message)

start_server = websockets.serve(echo, "127.0.0.1", 8765)

asyncio.get_event_loop().run_until_complete(start_server)
asyncio.get_event_loop().run_forever()

# WebSocket客户端代码
import asyncio
import websockets

async def client():
    async with websockets.connect("ws://127.0.0.1:8765") as websocket:
        await websocket.send("Hello, server!")
        response = await websocket.recv()
        print(f"收到服务器消息：{response}")

asyncio.get_event_loop().run_until_complete(client())

实战案例五：FTP客户端与服务器

FTP（文件传输协议）是一种用于文件传输的协议。以下是一个简单的FTP客户端与服务器通信的例子：

# FTP服务器端代码
import os
import ftplib

def start_ftp_server():
    server = ftplib.FTP()
    server.bind_addr = ('127.0.0.1', 21)
    server.set_pass('123456')
    server.listen()

    while True:
        conn, addr = server.accept()
        print(f"连接成功，客户端地址：{addr}")

        with conn:
            while True:
                cmd = conn.recv(1024).decode()
                if not cmd:
                    break
                if cmd.startswith('USER'):
                    conn.sendall(b"331 User name okay, need password.")
                elif cmd.startswith('PASS'):
                    conn.sendall(b"230 User logged in, proceed.")
                elif cmd.startswith('CWD'):
                    conn.sendall(b"250 CWD command successful.")
                elif cmd.startswith('LIST'):
                    files = os.listdir('.')
                    response = b"150 Here comes the directory listing:\n"
                    for file in files:
                        response += f"{file}\n".encode()
                    response += b"226 Directory send OK."
                    conn.sendall(response)
                elif cmd.startswith('RETR'):
                    file_name = cmd.split(' ')[1]
                    if os.path.exists(file_name):
                        with open(file_name, 'rb') as f:
                            conn.sendall(f.read())
                        conn.sendall(b"226 File send OK.")
                    else:
                        conn.sendall(b"450 File not found.")
                elif cmd.startswith('QUIT'):
                    conn.sendall(b"221 Goodbye.")
                    break

start_ftp_server()

# FTP客户端代码
import ftplib

def start_ftp_client():
    ftp = ftplib.FTP()
    ftp.connect('127.0.0.1', 21)
    ftp.login('user', '123456')

    # 列出文件
    ftp.retrlines('LIST')

    # 上传文件
    with open('example.txt', 'rb') as f:
        ftp.storbinary('STOR example.txt', f)

    # 下载文件
    with open('downloaded.txt', 'wb') as f:
        ftp.retrbinary('RETR example.txt', f.write)

    ftp.quit()

start_ftp_client()

实战案例六：SMTP客户端与服务器

SMTP（简单邮件传输协议）是一种用于电子邮件传输的协议。以下是一个简单的SMTP客户端与服务器通信的例子：

# SMTP服务器端代码
import smtplib
from email.mime.text import MIMEText
from email.header import Header

def send_email():
    sender = 'your_email@example.com'
    receivers = ['receiver1@example.com', 'receiver2@example.com']
    message = MIMEText('这是一封测试邮件', 'plain', 'utf-8')
    message['From'] = Header("菜鸟邮件", 'utf-8')
    message['To'] = Header("收件人", 'utf-8')
    message['Subject'] = Header("测试邮件", 'utf-8')

    try:
        smtp_obj = smtplib.SMTP('localhost')
        smtp_obj.sendmail(sender, receivers, message.as_string())
        print("邮件发送成功")
    except smtplib.SMTPException as e:
        print("无法发送邮件", e)

send_email()

# SMTP客户端代码
import smtplib
from email.mime.text import MIMEText
from email.header import Header

def send_email():
    sender = 'your_email@example.com'
    receivers = ['receiver1@example.com', 'receiver2@example.com']
    message = MIMEText('这是一封测试邮件', 'plain', 'utf-8')
    message['From'] = Header("菜鸟邮件", 'utf-8')
    message['To'] = Header("收件人", 'utf-8')
    message['Subject'] = Header("测试邮件", 'utf-8')

    try:
        smtp_obj = smtplib.SMTP('localhost')
        smtp_obj.sendmail(sender, receivers, message.as_string())
        print("邮件发送成功")
    except smtplib.SMTPException as e:
        print("无法发送邮件", e)

send_email()

实战案例七：DNS客户端与服务器

DNS（域名系统）是一种将域名转换为IP地址的系统。以下是一个简单的DNS客户端与服务器通信的例子：

# DNS服务器端代码
import socket

def start_dns_server():
    server_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    server_socket.bind(('127.0.0.1', 53))

    while True:
        data, addr = server_socket.recvfrom(1024)
        query = data.decode().split(' ')[1]
        print(f"收到DNS查询：{query}")

        # 查询本地hosts文件
        with open('/etc/hosts', 'r') as f:
            for line in f:
                if query in line:
                    ip = line.split()[0]
                    response = f"ANSWER {query} A {ip}\n"
                    server_socket.sendto(response.encode(), addr)
                    break
            else:
                response = f"ANSWER {query} NXDOMAIN\n"
                server_socket.sendto(response.encode(), addr)

start_dns_server()

# DNS客户端代码
import socket

def start_dns_client():
    query = 'www.example.com'
    server_ip = '127.0.0.1'
    server_port = 53

    # 创建socket对象
    client_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)

    # 发送DNS查询
    query_data = f"QUERY {query} A IN\n"
    client_socket.sendto(query_data.encode(), (server_ip, server_port))

    # 接收响应
    response, addr = client_socket.recvfrom(1024)
    print(f"收到DNS响应：{response.decode()}")

    # 关闭连接
    client_socket.close()

start_dns_client()

实战案例八：SSH客户端与服务器

SSH（安全外壳协议）是一种用于安全地访问远程服务器的方法。以下是一个简单的SSH客户端与服务器通信的例子：

# SSH服务器端代码
import paramiko

def start_ssh_server():
    # 创建SSH服务器
    server = paramiko.ServerInterface()

    # 创建socket对象
    server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server_socket.bind(('127.0.0.1', 22))
    server_socket.listen(5)

    # 接受客户端连接
    conn, addr = server_socket.accept()
    print(f"连接成功，客户端地址：{addr}")

    # 创建SSH客户端
    transport = paramiko.Transport(conn)
    transport.set_security_options(paramiko.DEFAULT_CLIENT_KEXS, paramiko.DEFAULT_CLIENT_CIPS, paramiko.DEFAULT_CLIENT_CPS)

    # 认证
    user_auth = paramiko.UserAuthPassword(username='user', password='123456')
    if transport.auth_password(username='user', password='123456'):
        print("认证成功")

    # 创建SSH会话
    chan = transport.open_session()
    chan.get_pty()
    chan.invoke_shell()

    # 发送数据
    chan.send("ls\n")

    # 接收数据
    while True:
        data = chan.recv(1024)
        if not data:
            break
        print(data.decode())

    # 关闭连接
    chan.close()
    transport.close()
    server_socket.close()

start_ssh_server()

# SSH客户端代码
import paramiko

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

    # 连接服务器
    client.connect('127.0.0.1', username='user', password='123456')

    # 创建SSH会话
    chan = client.get_transport().open_session()
    chan.get_pty()
    chan.invoke_shell()

    # 发送数据
    chan.send("ls\n")

    # 接收数据
    while True:
        data = chan.recv(1024)
        if not data:
            break
        print(data.decode())

    # 关闭连接
    chan.close()
    client.close()

start_ssh_client()

实战案例九：NTP客户端与服务器

NTP（网络时间协议）是一种用于同步网络设备时间的协议。以下是一个简单的NTP客户端与服务器通信的例子：

# NTP服务器端代码
import socket
import time

def start_ntp_server():
    # 创建socket对象
    server_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    server_socket.bind(('127.0.0.1', 123))

    while True:
        # 接收客户端数据
        data, addr = server_socket.recvfrom(1024)
        print(f"收到客户端数据：{data}")

        # 计算本地时间
        local_time = time.time()

        # 计算UTC时间
        utc_time = time.gmtime(local_time)

        # 构造NTP响应数据
        response = b'\x1b' + struct.pack("!12I", 0x4111cc33, 0, 0, 0, 0, 0, 0, 0, 0, 0, int(local_time * 1000000), 0)
        server_socket.sendto(response, addr)

start_ntp_server()

# NTP客户端代码
import socket
import struct

def start_ntp_client():
    # 创建socket对象
    client_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)

    # 发送NTP查询
    request = b'\x1b' + struct.pack("!12I", 0x4111cc33, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0)
    client_socket.sendto(request, ('pool.ntp.org', 123))

    # 接收NTP响应
    response, addr = client_socket.recvfrom(1024)
    print(f"收到NTP响应：{response}")

    # 计算UTC时间
    utc_time = struct.unpack("!12I", response)[10] / 1000000.0

    # 计算本地时间
    local_time = time.time()

    # 计算时间差
    time_diff = utc_time - local_time

    # 关闭连接
    client_socket.close()

    print(f"本地时间：{local_time}")
    print(f"UTC时间：{utc_time}")
    print(f"时间差：{time_diff}")

start_ntp_client()

实战案例十：SNMP客户端与服务器

SNMP（简单网络管理协议）是一种用于网络设备管理的协议。以下是一个简单的SNMP客户端与服务器通信的例子：

# SNMP服务器端代码
import sys
import socket
import struct
import time

# SNMP响应数据结构
class SNMPResponse:
    def __init__(self, error_status, error_index, varbind_list):
        self.error_status = error_status
        self.error_index = error_index
        self.varbind_list = varbind_list

    def pack(self):
        return struct.pack("!HH", self.error_status, self.error_index) + b''.join(
            struct.pack("!HH", 0, len(varbind)) + varbind.pack() for varbind in self.varbind_list)

# SNMP变量绑定数据结构
class SNMPVarBind:
    def __init__(self, name, value):
        self.name = name
        self.value = value

    def pack(self):
        return struct.pack("!HH", len(self.name), len(self.value)) + self.name.encode() + self.value

# SNMP服务器端函数
def start_snmp_server():
    # 创建socket对象
    server_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    server_socket.bind(('127.0.0.1', 161))

    while True:
        # 接收SNMP请求
        data, addr = server_socket.recvfrom(1024)
        print(f"收到SNMP请求：{data}")

        # 解析SNMP请求
        version, community, req_id, max_age, flags, req_type, varbind_list = parse_snmp_request(data)

        # 构造SNMP响应
        response = SNMPResponse(0, 0, [SNMPVarBind('1.3.6.1.2.1.1.1.0', 'Hello, world!')])

        # 发送SNMP响应
        server_socket.sendto(response.pack(), addr)

# 解析SNMP请求
def parse_snmp_request(data):
    # 解析版本、共同体、请求ID、最大存活时间、标志、请求类型
    version, community, req_id, max_age, flags, req_type = struct.unpack("!HHIIHH", data[:18])

    # 解析变量绑定列表
    varbind_list = []
    offset = 18
    while offset < len(data):
        name_len, value_len = struct.unpack("!HH", data[offset:offset+4])
        name = data[offset+4:offset+4+name_len].decode()
        value = data[offset+4+name_len:offset+4+name_len+value_len]
        varbind_list.append(SNMPVarBind(name, value))
        offset += 4 + name_len + value_len

    return version, community, req_id, max_age, flags, req_type, varbind_list

start_snmp_server()

”`python

SNMP客户端代码

import sys import socket import struct

SNMP请求数据结构

-- 展开阅读全文 --