如何高效爬取全球新聞網站 – 整合Scrapy、Selenium與Mediastack API實現自動化新聞采集
你可能也喜歡這些API文章!
身份驗證是接口（API）防止被調用的第一道防線。常見的身份驗證方式包括：
API Key 是一種簡單的身份驗證方式，每個合法用戶都會分配一個唯一的API Key 。調用接口時，需在請求頭或參數中附帶該 Key，服務器通過驗證 Key 的有效性來判斷調用者身份。
import requests
api_key = "your_api_key_here"
url = "https://api.example.com/data"
headers = {
"Authorization": f"Bearer {api_key}"
}
response = requests.get(url, headers=headers)
print(response.json())OAuth 2.0 是一種更為復雜的身份驗證協議，適用于需要更高安全性的場景。它通過授權碼、令牌等機制，確保只有經過授權的用戶才能訪問接口。
from oauthlib.oauth2 import BackendApplicationClient
from requests_oauthlib import OAuth2Session
client_id = "your_client_id"
client_secret = "your_client_secret"
token_url = "https://api.example.com/oauth/token"
client = BackendApplicationClient(client_id=client_id)
oauth = OAuth2Session(client=client)
token = oauth.fetch_token(token_url=token_url, client_id=client_id, client_secret=client_secret)
url = "https://api.example.com/data"
response = oauth.get(url)
print(response.json())身份驗證通過后，還需進一步控制調用者的操作權限。常見的權限控制方式包括：
RBAC 通過定義不同的角色，并為每個角色分配相應的權限，來實現精細化的權限控制。例如，管理員角色可以訪問所有接口，而普通用戶只能訪問部分接口。
from flask import Flask, request, jsonify
from functools import wraps
app = Flask(__name__)
# 模擬用戶角色和權限
users = {
"admin": {"role": "admin", "permissions": ["read", "write", "delete"]},
"user": {"role": "user", "permissions": ["read"]}
}
def requires_role(role):
def decorator(f):
@wraps(f)
def decorated_function(*args, **kwargs):
user_role = request.headers.get("X-User-Role")
if user_role != role:
return jsonify({"error": "Unauthorized"}), 403
return f(*args, **kwargs)
return decorated_function
return decorator
@app.route("/admin")
@requires_role("admin")
def admin():
return jsonify({"message": "Welcome, Admin!"})
@app.route("/user")
@requires_role("user")
def user():
return jsonify({"message": "Welcome, User!"})
if __name__ == "__main__":
app.run()ABAC 是一種更為靈活的權限控制方式，它通過評估調用者的屬性、資源屬性、環境條件等多個因素，動態決定是否允許訪問。
from flask import Flask, request, jsonify
from functools import wraps
app = Flask(__name__)
# 模擬用戶屬性
users = {
"alice": {"department": "HR", "location": "US"},
"bob": {"department": "IT", "location": "UK"}
}
def requires_permission(permission):
def decorator(f):
@wraps(f)
def decorated_function(*args, **kwargs):
user_id = request.headers.get("X-User-ID")
user = users.get(user_id)
if not user or permission not in user.get("permissions", []):
return jsonify({"error": "Unauthorized"}), 403
return f(*args, **kwargs)
return decorated_function
return decorator
@app.route("/resource")
@requires_permission("read")
def resource():
return jsonify({"message": "Access granted!"})
if __name__ == "__main__":
app.run()為了防止接口（API）被過度調用，頻率限制是必不可少的。常見的頻率限制方式包括：
令牌桶算法通過維護一個固定容量的令牌桶，每個請求需要消耗一個令牌。當令牌耗盡時，新的請求將被拒絕。
import time
from flask import Flask, request, jsonify
from functools import wraps
app = Flask(__name__)
# 令牌桶配置
bucket_capacity = 10
tokens = bucket_capacity
last_check = time.time()
def rate_limit(f):
@wraps(f)
def decorated_function(*args, **kwargs):
global tokens, last_check
now = time.time()
elapsed = now - last_check
last_check = now
# 每秒補充一個令牌
tokens += elapsed
if tokens > bucket_capacity:
tokens = bucket_capacity
if tokens < 1:
return jsonify({"error": "Rate limit exceeded"}), 429
tokens -= 1
return f(*args, **kwargs)
return decorated_function
@app.route("/api")
@rate_limit
def api():
return jsonify({"message": "Request successful!"})
if __name__ == "__main__":
app.run()漏桶算法通過固定速率處理請求，超出速率的請求將被丟棄或排隊等待。
import time
from flask import Flask, request, jsonify
from functools import wraps
app = Flask(__name__)
# 漏桶配置
leak_rate = 1 # 每秒處理1個請求
last_leak = time.time()
queue = []
def leaky_bucket(f):
@wraps(f)
def decorated_function(*args, **kwargs):
global last_leak, queue
now = time.time()
elapsed = now - last_leak
# 漏桶漏水
leaks = int(elapsed * leak_rate)
queue = queue[leaks:]
if len(queue) >= 10: # 隊列容量為10
return jsonify({"error": "Rate limit exceeded"}), 429
queue.append(now)
last_leak = now
return f(*args, **kwargs)
return decorated_function
@app.route("/api")
@leaky_bucket
def api():
return jsonify({"message": "Request successful!"})
if __name__ == "__main__":
app.run()為了保護傳輸數據的安全性，數據加密是必不可少的。常見的數據加密方式包括：
HTTPS 通過 SSL/TLS 協議對傳輸數據進行加密，防止數據在傳輸過程中被竊取或篡改。
import requests
url = "https://api.example.com/data"
response = requests.get(url)
print(response.json())JWT 是一種輕量級的身份驗證和數據交換格式，通過簽名確保數據的完整性和真實性。
import jwt
import datetime
# 生成 JWT
payload = {
"user_id": 123,
"exp": datetime.datetime.utcnow() + datetime.timedelta(hours=1)
}
secret_key = "your_secret_key"
token = jwt.encode(payload, secret_key, algorithm="HS256")
print(token)
# 驗證 JWT
try:
decoded = jwt.decode(token, secret_key, algorithms=["HS256"])
print(decoded)
except jwt.ExpiredSignatureError:
print("Token expired")
except jwt.InvalidTokenError:
print("Invalid token")單一的防護措施往往難以應對復雜的攻擊場景。因此，建議采用多層防護策略，結合身份驗證、權限控制、頻率限制和數據加密等多種手段，構建全方位的防護體系。
通過記錄接口調用的日志，可以及時發現異常行為，并采取相應的措施。建議對日志進行定期分析，識別潛在的安全威脅。
import logging
logging.basicConfig(filename="api.log", level=logging.INFO)
@app.route("/api")
@rate_limit
def api():
logging.info(f"API called by {request.remote_addr}")
return jsonify({"message": "Request successful!"})定期對接口（API）的安全性進行審計，發現并修復潛在的安全漏洞。建議采用自動化工具進行掃描，并結合人工審查，確保審計的全面性和準確性。
提高開發團隊的安全意識，定期進行安全培訓，確保每個成員都能掌握最新的安全技術和最佳實踐。
接口（API）防調用是一個復雜而重要的課題，需要從多個維度進行綜合考慮。通過身份驗證、權限控制、頻率限制和數據加密等技術手段，結合多層防護、日志監控、定期審計和安全培訓等最佳實踐，可以有效提升接口的安全性，保障系統的穩定運行和數據的安全傳輸。希望本文能為開發者提供有價值的參考，助力構建更加安全的互聯網環境。