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) -> datetime:

115 """

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

117 """

118 return ts.ToDatetime(tzinfo=UTC)

119

120

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

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

123 if isinstance(value, Timestamp):

124 return value.ToDatetime(timezone)

125

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

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

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

129

130 return value.astimezone(timezone)

131

132

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

134 """

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

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

137 """

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

139

140

141def now() -> datetime:

142 return datetime.now(tz=UTC)

143

144

145def minimum_allowed_birthdate() -> date:

146 """

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

148

149 This approximation works on leap days!

150 """

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

152

153

154def today() -> date:

155 """

156 Date only in UTC

157 """

158 return now().date()

159

160

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

162 """

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

164 """

165 return datetime.now(ZoneInfo(tz))

166

167

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

169 """

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

171 """

172 return now_in_timezone(tz).date()

173

174

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

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

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

178

179

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

181 """

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

183 """

184

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

186 if deg > ct:

187 deg -= adj

188 if deg < -ct:

189 deg += adj

190 return deg

191

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

193 if deg > ct:

194 deg = -deg + adj

195 if deg < -ct:

196 deg = -deg - adj

197 return deg

198

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

200 if deg > ct:

201 deg = deg % ct

202 if deg < -ct:

203 deg = deg % -ct

204 return deg

205

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

207 lng = __wrap_rem(lng)

208 lat = __wrap_rem(lat)

209 lng = __wrap_gen(lng, 180, 360)

210 lat = __wrap_flip(lat, 180, 180)

211 lat = __wrap_flip(lat, 90, 180)

212 if lng == -180:

213 lng = 180

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

215 lng = 0

216

217 return lat, lng

218

219

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

221 """

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

223 """

224 lat, lng = wrap_coordinate(lat, lng)

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

226

227

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

229 """

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

231 """

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

233

234

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

236 """

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

238 """

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

240

241

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

243 """

244 Turns GeoJSON to PostGIS geom data in EPSG4326

245 """

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

247

248

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

250 return func.ST_Multi(polygon)

251

252

253@overload

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

255@overload

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

257

258

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

260 """

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

262 """

263 if geom:

264 shp = to_shape(geom)

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

266 return shp.y, shp.x

267 else:

268 return None

269

270

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

272 """

273 Format the datetime for HTTP cookies

274 """

275 if not dt:

276 dt = now()

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

278

279

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

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

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

283 # tell the browser when to stop sending the cookie

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

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

286 cookie["domain"] = config.COOKIE_DOMAIN

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

288 cookie["path"] = "/"

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

290 # send only on requests from first-party domains

291 cookie["samesite"] = "Strict"

292 else:

293 # send on all requests, requires Secure

294 cookie["samesite"] = "None"

295 # only set cookie on HTTPS sites in production

296 cookie["secure"] = True

297 # not accessible from javascript

298 cookie["httponly"] = httponly

299

300 return cookie.OutputString()

301

302

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

304 """

305 Creates our session cookies.

306

307 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

308 """

309 return [

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

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

312 ]

313

314

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

316 return [

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

318 ]

319

320

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

322 """

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

324 """

325 if "cookie" not in headers:

326 return None

327

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

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

330

331 if not cookie:

332 return None

333

334 return cookie.value

335

336

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

338 """

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

340 """

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

342

343

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

345 """

346 Returns our user id cookie value or None

347 """

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

349

350

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

352 """

353 Returns language cookie or None

354 """

355 return _parse_cookie(headers, "NEXT_LOCALE")

356

357

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

359 """

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

361 """

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

363 return None

364

365 authorization = headers["authorization"]

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

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

368

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

370 return None

371

372 return authorization[7:]

373

374

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

376 cookie_value = _parse_cookie(headers, "sofa")

377 if not cookie_value:

378 return None

379

380 try:

381 decode_sofa(cookie_value)

382 return cookie_value

383 except Exception:

384 return None

385

386

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

388 sofa_value = encode_sofa(

389 create_sofa_id(),

390 internal_pb2.SofaPayload(

391 version=1,

392 created=Timestamp_from_datetime(now()),

393 ),

394 )

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

396

397

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

399 out = []

400 for item in list_:

401 if item not in out:

402 out.append(item)

403 return out

404

405

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

407 """

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

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

410

411 SET SESSION TIME ZONE 'America/New_York';

412

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

414

415 INSERT INTO tz_trouble(to_date, timezone) VALUES

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

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

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

419

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

421

422 The result is:

423

424 timezone

425 ------------------------

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

427 2021-03-19 19:00:00-04

428 2021-04-15 00:00:00-04

429 """

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

431

432

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

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

435

436

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

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

439

440

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

442 """

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

444

445 We pray to deities that this never changes

446 """

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

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

449

450

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

452 # see above comment

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

454

455

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

457 """

458 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

459 """

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

461

462

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

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

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

466 return x

467

468

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

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

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

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

473 return x