Coverage for app_modules/media_routes.py: 67%

1""" 

2media_routes.py 

3 

4Purpose: 

5 Serve user-owned images through signed URLs instead of exposing /static/User-photos directly. 

6 Requires the user to be authenticated and to own the camera the image belongs to. 

7 

8Routes: 

9 - GET /media/<token> 

10 

11Security: 

12 - Token is an HMAC of the normalized relative path under User-photos using SECRET_KEY 

13 - No expiry to remain stable; access still requires session auth + ownership check 

14 - Prevents enumeration of /static/User-photos and direct linking 

15""" 

16 

17import os 

18import base64 

19import hmac 

20import hashlib 

21from flask import Blueprint, abort, send_file, session, current_app 

22from .db import get_db 

23from .paths import STATIC_PATH 

24from .helpers import parse_ts_from_any, normalize_to_static_user_photos, resolve_share_token 

25 

26 

27bp = Blueprint('media_routes', __name__) 

28 

29 

30def _b64url_decode(data: str) -> bytes: 

31 padding = '=' * (-len(data) % 4) 

32 return base64.urlsafe_b64decode(data + padding) 

33 

34 

35@bp.route('/media/<path:token>') 

36def media_get(token: str): 

37 # Two token modes: session-required (stable) or public share (expiring) 

38 rel = None 

39 is_public = False 

40 

41 # Detect public share token: has two dots → three parts 

42 if token.count('.') == 2: 

43 resolved = resolve_share_token(token) 

44 if not resolved: 

45 abort(403) 

46 rel, exp = resolved 

47 # Expiry check 

48 import time as _t 

49 if _t.time() > exp: 

50 abort(410) # Gone 

51 is_public = True 

52 else: 

53 # Session token requires login and admin/user ownership 

54 user_id = session.get('user_id') 

55 if not user_id: 

56 abort(401) 

57 # Verify session token HMAC 

58 if '.' not in token: 

59 abort(400) 

60 b64_rel, b64_mac = token.split('.', 1) 

61 try: 

62 rel = _b64url_decode(b64_rel).decode('utf-8') 

63 except Exception: 

64 abort(400) 

65 rel = normalize_to_static_user_photos(rel) 

66 secret = (current_app.config.get('SECRET_KEY') or '').encode('utf-8') 

67 expected = hmac.new(secret, rel.encode('utf-8'), hashlib.sha256).digest() 

68 try: 

69 provided = _b64url_decode(b64_mac) 

70 except Exception: 

71 abort(400) 

72 if not hmac.compare_digest(expected, provided): 

73 abort(403) 

74 

75 # Ownership check by parsing camera id from filename (session mode only) 

76 _, cam = parse_ts_from_any(rel) 

77 if not cam: 

78 abort(404) 

79 db = get_db() 

80 if not is_public: 

81 owner = db.execute('SELECT 1 FROM cameras WHERE user_id=? AND camera_id=?', (session['user_id'], cam)).fetchone() 

82 if not owner: 

83 # Allow admins to view any user's media via admin UI 

84 row = db.execute('SELECT is_admin FROM users WHERE id=?', (session['user_id'],)).fetchone() 

85 if not row or int(row['is_admin'] or 0) != 1: 

86 abort(403) 

87 

88 # Serve file if under STATIC_PATH 

89 abs_path = os.path.join(STATIC_PATH, rel) 

90 real_abs = os.path.realpath(abs_path) 

91 if not real_abs.startswith(os.path.realpath(STATIC_PATH)): 

92 abort(400) 

93 if not os.path.exists(real_abs): 

94 abort(404) 

95 # Use send_file to stream; set caching headers short-lived 

96 resp = send_file(real_abs) 

97 resp.headers['Cache-Control'] = 'private, max-age=300' 

98 return resp 

99 

100