Coverage for app/backend/src/couchers/utils.py: 93%

1import http.cookies

2import re

3import typing

4from collections.abc import Mapping, Sequence

5from datetime import UTC, date, datetime, timedelta, tzinfo

6from email.utils import formatdate

7from typing import TYPE_CHECKING, Any, overload

8from zoneinfo import ZoneInfo

10import regex

11from geoalchemy2 import WKBElement, WKTElement

12from geoalchemy2.shape import from_shape, to_shape

13from google.protobuf.duration_pb2 import Duration

14from google.protobuf.timestamp_pb2 import Timestamp

15from shapely.geometry import Point, Polygon, shape

16from sqlalchemy import Function, cast

17from sqlalchemy.orm import Mapped

18from sqlalchemy.sql import func

19from sqlalchemy.types import DateTime

21from couchers.config import config

22from couchers.constants import (

23 EMAIL_REGEX,

24 PREFERRED_LANGUAGE_COOKIE_EXPIRY,

25 VALID_NAME_MAX_LENGTH,

26 VALID_NAME_MIN_LENGTH,

27 VALID_NAME_REGEX,

28)

29from couchers.crypto import (

30 create_sofa_id,

31 decode_sofa,

32 decrypt_page_token,

33 encode_sofa,

34 encrypt_page_token,

35)

36from couchers.proto.internal import internal_pb2

38_VALID_NAME_PATTERN = regex.compile(VALID_NAME_REGEX, regex.UNICODE)

40if TYPE_CHECKING:

41 from couchers.models import Geom

44# When a user logs in, they can basically input one of three things: user id, username, or email

45# These are three non-intersecting sets

46# * user_ids are numeric representations in base 10

47# * usernames are alphanumeric + underscores, at least 2 chars long, and don't start with a number,

48# and don't start or end with underscore

49# * emails are just whatever stack overflow says emails are ;)

52def is_valid_user_id(field: str) -> bool:

53 """

54 Checks if it's a string representing a base 10 integer not starting with 0

55 """

56 return re.match(r"[1-9][0-9]*$", field) is not None

59def is_valid_username(field: str) -> bool:

60 """

61 Checks if it's an alphanumeric + underscore, lowercase string, at least

62 two characters long, and starts with a letter, ends with alphanumeric

63 """

64 return re.match(r"[a-z][0-9a-z_]*[a-z0-9]$", field) is not None

67def is_valid_name(field: str) -> bool:

68 """

69 Checks that the name satisfies the same rules as the web frontend:

71 * only letters (any Unicode letter), whitespace, apostrophes, and hyphens

72 * no leading or trailing whitespace

73 * 2-100 characters

74 """

75 if len(field) > VALID_NAME_MAX_LENGTH or len(field) < VALID_NAME_MIN_LENGTH:

76 return False

78 return _VALID_NAME_PATTERN.fullmatch(field) is not None

81def is_valid_email(field: str) -> bool:

82 return re.match(EMAIL_REGEX, field) is not None

85def Timestamp_from_datetime(dt: datetime) -> Timestamp:

86 if dt.tzinfo is None: 86 ↛ 87line 86 didn't jump to line 87 because the condition on line 86 was never true

87 raise ValueError("Cannot convert a naive datetime to a timestamp.")

89 pb_ts = Timestamp()

90 pb_ts.FromDatetime(dt)

91 return pb_ts

94def Duration_from_timedelta(dt: timedelta) -> Duration:

95 pb_d = Duration()

96 pb_d.FromTimedelta(dt)

97 return pb_d

100def parse_date(date_str: str) -> date | None:

101 """

102 Parses a date-only string in the format "YYYY-MM-DD" returning None if it fails

103 """

104 try:

105 return date.fromisoformat(date_str)

106 except ValueError:

107 return None

108

109

110def date_to_api(date_obj: date) -> str:

111 return date_obj.isoformat()

112

113

114def to_aware_datetime(ts: Timestamp, timezone: str | tzinfo = UTC) -> datetime:

115 """

116 Turns a protobuf Timestamp object into a timezone-aware datetime

117 """

118 if isinstance(timezone, str):

119 timezone = ZoneInfo(timezone)

120 return ts.ToDatetime(tzinfo=timezone)

121

122

123def to_timezone(value: Timestamp | datetime, timezone: tzinfo) -> datetime:

124 """Returns an instant in time as a datetime in a given timezone."""

125 if isinstance(value, Timestamp):

126 return value.ToDatetime(timezone)

127

128 if value.tzinfo is None: 128 ↛ 130line 128 didn't jump to line 130 because the condition on line 128 was never true

129 # A naive datetime does not represent a point in time.

130 raise ValueError("Cannot convert a naive datetime to a timezone.")

131

132 return value.astimezone(timezone)

133

134

135def datetime_to_iso8601_local(value: datetime) -> str:

136 """

137 Gets a local ISO 8601 representation of a datetime, without timezone information.

138 This loses information and requires parsers to assume a timezone, so use with care.

139 """

140 return value.replace(tzinfo=None).isoformat()

141

142

143def now() -> datetime:

144 return datetime.now(tz=UTC)

145

146

147def minimum_allowed_birthdate() -> date:

148 """

149 Most recent birthdate allowed to register (must be 18 years minimum)

150

151 This approximation works on leap days!

152 """

153 return today() - timedelta(days=365.25 * 18)

154

155

156def today() -> date:

157 """

158 Date only in UTC

159 """

160 return now().date()

161

162

163def now_in_timezone(tz: str) -> datetime:

164 """

165 tz should be tzdata identifier, e.g. America/New_York

166 """

167 return datetime.now(ZoneInfo(tz))

168

169

170def today_in_timezone(tz: str) -> date:

171 """

172 tz should be tzdata identifier, e.g. America/New_York

173 """

174 return now_in_timezone(tz).date()

175

176

177# Note: be very careful with ordering of lat/lng!

178# In a lot of cases they come as (lng, lat), but us humans tend to use them from GPS as (lat, lng)...

179# When entering as EPSG4326, we also need it in (lng, lat)

180

181

182def wrap_coordinate(lat: float, lng: float) -> tuple[float, float]:

183 """

184 Wraps (lat, lng) point in the EPSG4326 format

185 """

186

187 def __wrap_gen(deg: float, ct: float, adj: float) -> float:

188 if deg > ct:

189 deg -= adj

190 if deg < -ct:

191 deg += adj

192 return deg

193

194 def __wrap_flip(deg: float, ct: float, adj: float) -> float:

195 if deg > ct:

196 deg = -deg + adj

197 if deg < -ct:

198 deg = -deg - adj

199 return deg

200

201 def __wrap_rem(deg: float, ct: float = 360) -> float:

202 if deg > ct:

203 deg = deg % ct

204 if deg < -ct:

205 deg = deg % -ct

206 return deg

207

208 if lng < -180 or lng > 180 or lat < -90 or lat > 90:

209 lng = __wrap_rem(lng)

210 lat = __wrap_rem(lat)

211 lng = __wrap_gen(lng, 180, 360)

212 lat = __wrap_flip(lat, 180, 180)

213 lat = __wrap_flip(lat, 90, 180)

214 if lng == -180:

215 lng = 180

216 if lng == -360: 216 ↛ 217line 216 didn't jump to line 217 because the condition on line 216 was never true

217 lng = 0

218

219 return lat, lng

220

221

222def create_coordinate(lat: float, lng: float) -> WKBElement:

223 """

224 Creates a WKT point from a (lat, lng) tuple in EPSG4326 coordinate system (normal GPS-coordinates)

225 """

226 lat, lng = wrap_coordinate(lat, lng)

227 return from_shape(Point(lng, lat), srid=4326)

228

229

230def create_polygon_lat_lng(points: list[list[float]]) -> WKBElement:

231 """

232 Creates a EPSG4326 WKT polygon from a list of (lat, lng) tuples

233 """

234 return from_shape(Polygon([(lng, lat) for (lat, lng) in points]), srid=4326)

235

236

237def create_polygon_lng_lat(points: list[list[float]]) -> WKBElement:

238 """

239 Creates a EPSG4326 WKT polygon from a list of (lng, lat) tuples

240 """

241 return from_shape(Polygon(points), srid=4326)

242

243

244def geojson_to_geom(geojson: dict[str, Any]) -> WKBElement:

245 """

246 Turns GeoJSON to PostGIS geom data in EPSG4326

247 """

248 return from_shape(shape(geojson), srid=4326)

249

250

251def to_multi(polygon: WKBElement) -> Function[Any]:

252 return func.ST_Multi(polygon)

253

254

255@overload

256def get_coordinates(geom: WKBElement | WKTElement) -> tuple[float, float]: ...

257@overload

258def get_coordinates(geom: None) -> None: ...

259

260

261def get_coordinates(geom: WKBElement | WKTElement | None) -> tuple[float, float] | None:

262 """

263 Returns EPSG4326 (lat, lng) pair for a given WKT geom point or None if the input is not truthy

264 """

265 if geom:

266 shp = to_shape(geom)

267 # note the funniness with 4326 normally being (x, y) = (lng, lat)

268 return shp.y, shp.x

269 else:

270 return None

271

272

273def http_date(dt: datetime | None = None) -> str:

274 """

275 Format the datetime for HTTP cookies

276 """

277 if not dt:

278 dt = now()

279 return formatdate(dt.timestamp(), usegmt=True)

280

281

282def _create_tasty_cookie(name: str, value: Any, expiry: datetime, httponly: bool) -> str:

283 cookie: http.cookies.Morsel[str] = http.cookies.Morsel()

284 cookie.set(name, str(value), str(value))

285 # tell the browser when to stop sending the cookie

286 cookie["expires"] = http_date(expiry)

287 # restrict to our domain, note if there's no domain, it won't include subdomains

288 cookie["domain"] = config.COOKIE_DOMAIN

289 # path so that it's accessible for all API requests, otherwise defaults to something like /org.couchers.auth/

290 cookie["path"] = "/"

291 if config.DEV: 291 ↛ 296line 291 didn't jump to line 296 because the condition on line 291 was always true

292 # send only on requests from first-party domains

293 cookie["samesite"] = "Strict"

294 else:

295 # send on all requests, requires Secure

296 cookie["samesite"] = "None"

297 # only set cookie on HTTPS sites in production

298 cookie["secure"] = True

299 # not accessible from javascript

300 cookie["httponly"] = httponly

301

302 return cookie.OutputString()

303

304

305def create_session_cookies(token: str, user_id: str | int, expiry: datetime) -> list[str]:

306 """

307 Creates our session cookies.

308

309 We have two: the secure session token (in couchers-sesh) that's inaccessible to javascript, and the user id (in couchers-user-id) which the javascript frontend can access, so that it knows when it's logged in/out

310 """

311 return [

312 _create_tasty_cookie("couchers-sesh", token, expiry, httponly=True),

313 _create_tasty_cookie("couchers-user-id", user_id, expiry, httponly=False),

314 ]

315

316

317def create_lang_cookie(lang: str) -> list[str]:

318 return [

319 _create_tasty_cookie("NEXT_LOCALE", lang, expiry=(now() + PREFERRED_LANGUAGE_COOKIE_EXPIRY), httponly=False)

320 ]

321

322

323def _parse_cookie(headers: Mapping[str, str | bytes], cookie_name: str) -> str | None:

324 """

325 Helper to parse a cookie value from headers by name, returning None if not found.

326 """

327 if "cookie" not in headers:

328 return None

329

330 cookie_str = typing.cast(str, headers["cookie"])

331 cookie = http.cookies.SimpleCookie(cookie_str).get(cookie_name)

332

333 if not cookie:

334 return None

335

336 return cookie.value

337

338

339def parse_session_cookie(headers: Mapping[str, str | bytes]) -> str | None:

340 """

341 Returns our session cookie value (aka token) or None

342 """

343 return _parse_cookie(headers, "couchers-sesh")

344

345

346def parse_user_id_cookie(headers: Mapping[str, str | bytes]) -> str | None:

347 """

348 Returns our user id cookie value or None

349 """

350 return _parse_cookie(headers, "couchers-user-id")

351

352

353def parse_ui_lang_cookie(headers: Mapping[str, str | bytes]) -> str | None:

354 """

355 Returns language cookie or None

356 """

357 return _parse_cookie(headers, "NEXT_LOCALE")

358

359

360def parse_api_key(headers: Mapping[str, str | bytes]) -> str | None:

361 """

362 Returns a bearer token (API key) from the `authorization` header, or None if invalid/not present

363 """

364 if "authorization" not in headers: 364 ↛ 365line 364 didn't jump to line 365 because the condition on line 364 was never true

365 return None

366

367 authorization = headers["authorization"]

368 if isinstance(authorization, bytes): 368 ↛ 369line 368 didn't jump to line 369 because the condition on line 368 was never true

369 authorization = authorization.decode("utf-8")

370

371 if not authorization.startswith("Bearer "):

372 return None

373

374 return authorization[7:]

375

376

377def parse_sofa_cookie(headers: Mapping[str, str | bytes]) -> str | None:

378 cookie_value = _parse_cookie(headers, "sofa")

379 if not cookie_value:

380 return None

381

382 try:

383 decode_sofa(cookie_value)

384 return cookie_value

385 except Exception:

386 return None

387

388

389def generate_sofa_cookie() -> tuple[str, str]:

390 sofa_value = encode_sofa(

391 create_sofa_id(),

392 internal_pb2.SofaPayload(

393 version=1,

394 created=Timestamp_from_datetime(now()),

395 ),

396 )

397 return sofa_value, _create_tasty_cookie("sofa", sofa_value, now() + timedelta(days=10000), httponly=True)

398

399

400def remove_duplicates_retain_order[T](list_: Sequence[T]) -> list[T]:

401 out = []

402 for item in list_:

403 if item not in out:

404 out.append(item)

405 return out

406

407

408def date_in_timezone(date_: Mapped[date | None], timezone: str) -> Function[Any]:

409 """

410 Given a naive postgres date object (postgres doesn't have tzd dates), returns a timezone-aware timestamp for the

411 start of that date in that timezone. E.g., if postgres is in 'America/New_York',

412

413 SET SESSION TIME ZONE 'America/New_York';

414

415 CREATE TABLE tz_trouble (to_date date, timezone text);

416

417 INSERT INTO tz_trouble(to_date, timezone) VALUES

418 ('2021-03-10'::date, 'Australia/Sydney'),

419 ('2021-03-20'::date, 'Europe/Berlin'),

420 ('2021-04-15'::date, 'America/New_York');

421

422 SELECT timezone(timezone, to_date::timestamp) FROM tz_trouble;

423

424 The result is:

425

426 timezone

427 ------------------------

428 2021-03-09 08:00:00-05

429 2021-03-19 19:00:00-04

430 2021-04-15 00:00:00-04

431 """

432 return func.timezone(timezone, cast(date_, DateTime(timezone=False)))

433

434

435def millis_from_dt(dt: datetime) -> int:

436 return round(1000 * dt.timestamp())

437

438

439def dt_from_millis(millis: int) -> datetime:

440 return datetime.fromtimestamp(millis / 1000, tz=UTC)

441

442

443def dt_to_page_token(dt: datetime) -> str:

444 """

445 Python has datetime resolution equal to 1 micro, as does postgres

446

447 We pray to deities that this never changes

448 """

449 assert datetime.resolution == timedelta(microseconds=1)

450 return encrypt_page_token(str(round(1_000_000 * dt.timestamp())))

451

452

453def dt_from_page_token(page_token: str) -> datetime:

454 # see above comment

455 return datetime.fromtimestamp(int(decrypt_page_token(page_token)) / 1_000_000, tz=UTC)

456

457

458def last_active_coarsen(dt: datetime) -> datetime:

459 """

460 Coarsens a "last active" time to the accuracy we use for last active times, currently to the last hour, e.g. if the current time is 27th June 2021, 16:53 UTC, this returns 27th June 2021, 16:00 UTC

461 """

462 return dt.replace(minute=0, second=0, microsecond=0)

463

464

465def not_none[T](x: T | None) -> T:

466 if x is None: 466 ↛ 467line 466 didn't jump to line 467 because the condition on line 466 was never true

467 raise ValueError("Expected a value but got None")

468 return x

469

470

471def is_geom(x: Geom | None) -> Geom:

472 """not_none does not work with unions."""

473 if x is None: 473 ↛ 474line 473 didn't jump to line 474 because the condition on line 473 was never true

474 raise ValueError("Expected a Geom but got None")

475 return x