From f6c428d5af59c91732925ba08967c9b01f47989c Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 17 Jul 2026 16:51:50 +0200 Subject: [PATCH] Show a QR code for device-code login (text dialog as fallback) The device-code login now displays a scannable QR code of the Twitch activation URL alongside the short user code, so viewers can sign in by pointing their phone camera at the TV instead of typing a URL. Both the regular and the private (Turbo/ad-free) login get this, since they share do_device_login(). The QR is rendered dependency-free: a vendored QR generator (Project Nayuki, MIT) produces the module matrix and a tiny zlib-based PNG writer emits the image, so no PIL/Pillow or other binary library is required. It is shown in a lightweight WindowDialog that polls in a background thread. Anything that can go wrong with the QR path (rendering failure, an image-less skin, an exception building the window) falls back to the previous text-only DialogProgress flow, which is preserved unchanged, so login still works everywhere. Polling, slow_down handling, cancel, expiry and Kodi-shutdown behaviour are shared between both paths. Co-Authored-By: Claude Fable 5 --- .../resource.language.en_gb/strings.po | 12 + resources/lib/twitch_addon/addon/qr.py | 82 ++ resources/lib/twitch_addon/addon/strings.py | 3 + .../lib/twitch_addon/addon/vendor/__init__.py | 2 + .../twitch_addon/addon/vendor/qrcodegen.py | 909 ++++++++++++++++++ .../lib/twitch_addon/routes/device_login.py | 243 ++++- 6 files changed, 1223 insertions(+), 28 deletions(-) create mode 100644 resources/lib/twitch_addon/addon/qr.py create mode 100644 resources/lib/twitch_addon/addon/vendor/__init__.py create mode 100644 resources/lib/twitch_addon/addon/vendor/qrcodegen.py diff --git a/resources/language/resource.language.en_gb/strings.po b/resources/language/resource.language.en_gb/strings.po index 2d54d5c9..013dc33d 100644 --- a/resources/language/resource.language.en_gb/strings.po +++ b/resources/language/resource.language.en_gb/strings.po @@ -1073,3 +1073,15 @@ msgstr "" msgctxt "#30332" msgid "Helix API" msgstr "" + +msgctxt "#30334" +msgid "Scan the QR code, or go to [B]%s[/B]" +msgstr "" + +msgctxt "#30335" +msgid "Enter this code: [B]%s[/B]" +msgstr "" + +msgctxt "#30336" +msgid "Waiting for authorization... (press Back to cancel)" +msgstr "" diff --git a/resources/lib/twitch_addon/addon/qr.py b/resources/lib/twitch_addon/addon/qr.py new file mode 100644 index 00000000..1a1adeed --- /dev/null +++ b/resources/lib/twitch_addon/addon/qr.py @@ -0,0 +1,82 @@ +# -*- coding: utf-8 -*- +""" + + Copyright (C) 2024-2025 Twitch-on-Kodi + + This file is part of Twitch-on-Kodi (plugin.video.twitch) + + SPDX-License-Identifier: GPL-3.0-only + See LICENSES/GPL-3.0-only for more information. + + Dependency-free QR code -> PNG rendering used by the device-code login flow. + Uses the vendored Project Nayuki QR generator (MIT) for the module matrix and + a tiny zlib-based PNG writer so no binary image library (PIL/Pillow) is needed. +""" + +import struct +import zlib + +from .vendor.qrcodegen import QrCode + + +def _write_grayscale_png(file_path, width, height, rows): + """Write an 8-bit grayscale PNG. ``rows`` is an iterable of ``width``-length + bytes objects (one byte per pixel, 0=black .. 255=white).""" + + def chunk(tag, data): + out = struct.pack('>I', len(data)) + tag + data + out += struct.pack('>I', zlib.crc32(tag + data) & 0xFFFFFFFF) + return out + + raw = bytearray() + for row in rows: + raw.append(0) # filter type 0 (None) for each scanline + raw.extend(row) + + ihdr = struct.pack('>IIBBBBB', width, height, 8, 0, 0, 0, 0) # 8-bit grayscale + png = b'\x89PNG\r\n\x1a\n' + png += chunk(b'IHDR', ihdr) + png += chunk(b'IDAT', zlib.compress(bytes(raw), 9)) + png += chunk(b'IEND', b'') + + with open(file_path, 'wb') as handle: + handle.write(png) + return file_path + + +def generate_qr_png(data, file_path, scale=8, border=4, ecc=None): + """Render ``data`` as a QR code PNG at ``file_path``. + + :param scale: pixels per QR module + :param border: quiet-zone width in modules (the spec requires >= 4) + :returns: ``file_path`` on success + """ + if ecc is None: + ecc = QrCode.Ecc.MEDIUM + qr = QrCode.encode_text(data, ecc) + size = qr.get_size() + total = size + 2 * border + width = height = total * scale + + rows = [] + for ty in range(total): + line = bytearray(width) + for tx in range(total): + mx = tx - border + my = ty - border + dark = (0 <= mx < size) and (0 <= my < size) and qr.get_module(mx, my) + value = 0 if dark else 255 + base = tx * scale + for sx in range(scale): + line[base + sx] = value + packed = bytes(line) + for _ in range(scale): + rows.append(packed) + + return _write_grayscale_png(file_path, width, height, rows) + + +def generate_solid_png(file_path, width, height, value=18): + """Write a solid grayscale rectangle, used as a backdrop panel behind the QR.""" + row = bytes([value]) * width + return _write_grayscale_png(file_path, width, height, (row for _ in range(height))) diff --git a/resources/lib/twitch_addon/addon/strings.py b/resources/lib/twitch_addon/addon/strings.py index c93023e0..8af07ff5 100644 --- a/resources/lib/twitch_addon/addon/strings.py +++ b/resources/lib/twitch_addon/addon/strings.py @@ -125,6 +125,9 @@ 'device_login_failed': 30304, 'device_login_cancelled': 30305, 'device_login_unsupported': 30306, + 'device_login_scan': 30334, + 'device_login_code': 30335, + 'device_login_waiting_cancel': 30336, 'oauth_refresh_needs_public_client': 30310, 'device_login_private': 30311, 'device_login_private_instructions': 30312, diff --git a/resources/lib/twitch_addon/addon/vendor/__init__.py b/resources/lib/twitch_addon/addon/vendor/__init__.py new file mode 100644 index 00000000..34b3c3c0 --- /dev/null +++ b/resources/lib/twitch_addon/addon/vendor/__init__.py @@ -0,0 +1,2 @@ +# -*- coding: utf-8 -*- +# Vendored third-party libraries. diff --git a/resources/lib/twitch_addon/addon/vendor/qrcodegen.py b/resources/lib/twitch_addon/addon/vendor/qrcodegen.py new file mode 100644 index 00000000..5d84c39e --- /dev/null +++ b/resources/lib/twitch_addon/addon/vendor/qrcodegen.py @@ -0,0 +1,909 @@ +# +# QR Code generator library (Python) +# +# Copyright (c) Project Nayuki. (MIT License) +# https://www.nayuki.io/page/qr-code-generator-library +# +# Permission is hereby granted, free of charge, to any person obtaining a copy of +# this software and associated documentation files (the "Software"), to deal in +# the Software without restriction, including without limitation the rights to +# use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of +# the Software, and to permit persons to whom the Software is furnished to do so, +# subject to the following conditions: +# - The above copyright notice and this permission notice shall be included in +# all copies or substantial portions of the Software. +# - The Software is provided "as is", without warranty of any kind, express or +# implied, including but not limited to the warranties of merchantability, +# fitness for a particular purpose and noninfringement. In no event shall the +# authors or copyright holders be liable for any claim, damages or other +# liability, whether in an action of contract, tort or otherwise, arising from, +# out of or in connection with the Software or the use or other dealings in the +# Software. +# + +from __future__ import annotations +import collections, itertools, re +from collections.abc import Sequence +from typing import Callable, Dict, List, Optional, Tuple, Union + + +# ---- QR Code symbol class ---- + +class QrCode: + """A QR Code symbol, which is a type of two-dimension barcode. + Invented by Denso Wave and described in the ISO/IEC 18004 standard. + Instances of this class represent an immutable square grid of dark and light cells. + The class provides static factory functions to create a QR Code from text or binary data. + The class covers the QR Code Model 2 specification, supporting all versions (sizes) + from 1 to 40, all 4 error correction levels, and 4 character encoding modes. + + Ways to create a QR Code object: + - High level: Take the payload data and call QrCode.encode_text() or QrCode.encode_binary(). + - Mid level: Custom-make the list of segments and call QrCode.encode_segments(). + - Low level: Custom-make the array of data codeword bytes (including + segment headers and final padding, excluding error correction codewords), + supply the appropriate version number, and call the QrCode() constructor. + (Note that all ways require supplying the desired error correction level.)""" + + # ---- Static factory functions (high level) ---- + + @staticmethod + def encode_text(text: str, ecl: QrCode.Ecc) -> QrCode: + """Returns a QR Code representing the given Unicode text string at the given error correction level. + As a conservative upper bound, this function is guaranteed to succeed for strings that have 738 or fewer + Unicode code points (not UTF-16 code units) if the low error correction level is used. The smallest possible + QR Code version is automatically chosen for the output. The ECC level of the result may be higher than the + ecl argument if it can be done without increasing the version.""" + segs: List[QrSegment] = QrSegment.make_segments(text) + return QrCode.encode_segments(segs, ecl) + + + @staticmethod + def encode_binary(data: Union[bytes,Sequence[int]], ecl: QrCode.Ecc) -> QrCode: + """Returns a QR Code representing the given binary data at the given error correction level. + This function always encodes using the binary segment mode, not any text mode. The maximum number of + bytes allowed is 2953. The smallest possible QR Code version is automatically chosen for the output. + The ECC level of the result may be higher than the ecl argument if it can be done without increasing the version.""" + return QrCode.encode_segments([QrSegment.make_bytes(data)], ecl) + + + # ---- Static factory functions (mid level) ---- + + @staticmethod + def encode_segments(segs: Sequence[QrSegment], ecl: QrCode.Ecc, minversion: int = 1, maxversion: int = 40, mask: int = -1, boostecl: bool = True) -> QrCode: + """Returns a QR Code representing the given segments with the given encoding parameters. + The smallest possible QR Code version within the given range is automatically + chosen for the output. Iff boostecl is true, then the ECC level of the result + may be higher than the ecl argument if it can be done without increasing the + version. The mask number is either between 0 to 7 (inclusive) to force that + mask, or -1 to automatically choose an appropriate mask (which may be slow). + This function allows the user to create a custom sequence of segments that switches + between modes (such as alphanumeric and byte) to encode text in less space. + This is a mid-level API; the high-level API is encode_text() and encode_binary().""" + + if not (QrCode.MIN_VERSION <= minversion <= maxversion <= QrCode.MAX_VERSION) or not (-1 <= mask <= 7): + raise ValueError("Invalid value") + + # Find the minimal version number to use + for version in range(minversion, maxversion + 1): + datacapacitybits: int = QrCode._get_num_data_codewords(version, ecl) * 8 # Number of data bits available + datausedbits: Optional[int] = QrSegment.get_total_bits(segs, version) + if (datausedbits is not None) and (datausedbits <= datacapacitybits): + break # This version number is found to be suitable + if version >= maxversion: # All versions in the range could not fit the given data + msg: str = "Segment too long" + if datausedbits is not None: + msg = f"Data length = {datausedbits} bits, Max capacity = {datacapacitybits} bits" + raise DataTooLongError(msg) + assert datausedbits is not None + + # Increase the error correction level while the data still fits in the current version number + for newecl in (QrCode.Ecc.MEDIUM, QrCode.Ecc.QUARTILE, QrCode.Ecc.HIGH): # From low to high + if boostecl and (datausedbits <= QrCode._get_num_data_codewords(version, newecl) * 8): + ecl = newecl + + # Concatenate all segments to create the data bit string + bb = _BitBuffer() + for seg in segs: + bb.append_bits(seg.get_mode().get_mode_bits(), 4) + bb.append_bits(seg.get_num_chars(), seg.get_mode().num_char_count_bits(version)) + bb.extend(seg._bitdata) + assert len(bb) == datausedbits + + # Add terminator and pad up to a byte if applicable + datacapacitybits = QrCode._get_num_data_codewords(version, ecl) * 8 + assert len(bb) <= datacapacitybits + bb.append_bits(0, min(4, datacapacitybits - len(bb))) + bb.append_bits(0, -len(bb) % 8) # Note: Python's modulo on negative numbers behaves better than C family languages + assert len(bb) % 8 == 0 + + # Pad with alternating bytes until data capacity is reached + for padbyte in itertools.cycle((0xEC, 0x11)): + if len(bb) >= datacapacitybits: + break + bb.append_bits(padbyte, 8) + + # Pack bits into bytes in big endian + datacodewords = bytearray([0] * (len(bb) // 8)) + for (i, bit) in enumerate(bb): + datacodewords[i >> 3] |= bit << (7 - (i & 7)) + + # Create the QR Code object + return QrCode(version, ecl, datacodewords, mask) + + + # ---- Private fields ---- + + # The version number of this QR Code, which is between 1 and 40 (inclusive). + # This determines the size of this barcode. + _version: int + + # The width and height of this QR Code, measured in modules, between + # 21 and 177 (inclusive). This is equal to version * 4 + 17. + _size: int + + # The error correction level used in this QR Code. + _errcorlvl: QrCode.Ecc + + # The index of the mask pattern used in this QR Code, which is between 0 and 7 (inclusive). + # Even if a QR Code is created with automatic masking requested (mask = -1), + # the resulting object still has a mask value between 0 and 7. + _mask: int + + # The modules of this QR Code (False = light, True = dark). + # Immutable after constructor finishes. Accessed through get_module(). + _modules: List[List[bool]] + + # Indicates function modules that are not subjected to masking. Discarded when constructor finishes. + _isfunction: List[List[bool]] + + + # ---- Constructor (low level) ---- + + def __init__(self, version: int, errcorlvl: QrCode.Ecc, datacodewords: Union[bytes,Sequence[int]], msk: int) -> None: + """Creates a new QR Code with the given version number, + error correction level, data codeword bytes, and mask number. + This is a low-level API that most users should not use directly. + A mid-level API is the encode_segments() function.""" + + # Check scalar arguments and set fields + if not (QrCode.MIN_VERSION <= version <= QrCode.MAX_VERSION): + raise ValueError("Version value out of range") + if not (-1 <= msk <= 7): + raise ValueError("Mask value out of range") + + self._version = version + self._size = version * 4 + 17 + self._errcorlvl = errcorlvl + + # Initialize both grids to be size*size arrays of Boolean false + self._modules = [[False] * self._size for _ in range(self._size)] # Initially all light + self._isfunction = [[False] * self._size for _ in range(self._size)] + + # Compute ECC, draw modules + self._draw_function_patterns() + allcodewords: bytes = self._add_ecc_and_interleave(bytearray(datacodewords)) + self._draw_codewords(allcodewords) + + # Do masking + if msk == -1: # Automatically choose best mask + minpenalty: int = 1 << 32 + for i in range(8): + self._apply_mask(i) + self._draw_format_bits(i) + penalty = self._get_penalty_score() + if penalty < minpenalty: + msk = i + minpenalty = penalty + self._apply_mask(i) # Undoes the mask due to XOR + assert 0 <= msk <= 7 + self._mask = msk + self._apply_mask(msk) # Apply the final choice of mask + self._draw_format_bits(msk) # Overwrite old format bits + + del self._isfunction + + + # ---- Accessor methods ---- + + def get_version(self) -> int: + """Returns this QR Code's version number, in the range [1, 40].""" + return self._version + + def get_size(self) -> int: + """Returns this QR Code's size, in the range [21, 177].""" + return self._size + + def get_error_correction_level(self) -> QrCode.Ecc: + """Returns this QR Code's error correction level.""" + return self._errcorlvl + + def get_mask(self) -> int: + """Returns this QR Code's mask, in the range [0, 7].""" + return self._mask + + def get_module(self, x: int, y: int) -> bool: + """Returns the color of the module (pixel) at the given coordinates, which is False + for light or True for dark. The top left corner has the coordinates (x=0, y=0). + If the given coordinates are out of bounds, then False (light) is returned.""" + return (0 <= x < self._size) and (0 <= y < self._size) and self._modules[y][x] + + + # ---- Private helper methods for constructor: Drawing function modules ---- + + def _draw_function_patterns(self) -> None: + """Reads this object's version field, and draws and marks all function modules.""" + # Draw horizontal and vertical timing patterns + for i in range(self._size): + self._set_function_module(6, i, i % 2 == 0) + self._set_function_module(i, 6, i % 2 == 0) + + # Draw 3 finder patterns (all corners except bottom right; overwrites some timing modules) + self._draw_finder_pattern(3, 3) + self._draw_finder_pattern(self._size - 4, 3) + self._draw_finder_pattern(3, self._size - 4) + + # Draw numerous alignment patterns + alignpatpos: List[int] = self._get_alignment_pattern_positions() + numalign: int = len(alignpatpos) + skips: Sequence[Tuple[int,int]] = ((0, 0), (0, numalign - 1), (numalign - 1, 0)) + for i in range(numalign): + for j in range(numalign): + if (i, j) not in skips: # Don't draw on the three finder corners + self._draw_alignment_pattern(alignpatpos[i], alignpatpos[j]) + + # Draw configuration data + self._draw_format_bits(0) # Dummy mask value; overwritten later in the constructor + self._draw_version() + + + def _draw_format_bits(self, mask: int) -> None: + """Draws two copies of the format bits (with its own error correction code) + based on the given mask and this object's error correction level field.""" + # Calculate error correction code and pack bits + data: int = self._errcorlvl.formatbits << 3 | mask # errCorrLvl is uint2, mask is uint3 + rem: int = data + for _ in range(10): + rem = (rem << 1) ^ ((rem >> 9) * 0x537) + bits: int = (data << 10 | rem) ^ 0x5412 # uint15 + assert bits >> 15 == 0 + + # Draw first copy + for i in range(0, 6): + self._set_function_module(8, i, _get_bit(bits, i)) + self._set_function_module(8, 7, _get_bit(bits, 6)) + self._set_function_module(8, 8, _get_bit(bits, 7)) + self._set_function_module(7, 8, _get_bit(bits, 8)) + for i in range(9, 15): + self._set_function_module(14 - i, 8, _get_bit(bits, i)) + + # Draw second copy + for i in range(0, 8): + self._set_function_module(self._size - 1 - i, 8, _get_bit(bits, i)) + for i in range(8, 15): + self._set_function_module(8, self._size - 15 + i, _get_bit(bits, i)) + self._set_function_module(8, self._size - 8, True) # Always dark + + + def _draw_version(self) -> None: + """Draws two copies of the version bits (with its own error correction code), + based on this object's version field, iff 7 <= version <= 40.""" + if self._version < 7: + return + + # Calculate error correction code and pack bits + rem: int = self._version # version is uint6, in the range [7, 40] + for _ in range(12): + rem = (rem << 1) ^ ((rem >> 11) * 0x1F25) + bits: int = self._version << 12 | rem # uint18 + assert bits >> 18 == 0 + + # Draw two copies + for i in range(18): + bit: bool = _get_bit(bits, i) + a: int = self._size - 11 + i % 3 + b: int = i // 3 + self._set_function_module(a, b, bit) + self._set_function_module(b, a, bit) + + + def _draw_finder_pattern(self, x: int, y: int) -> None: + """Draws a 9*9 finder pattern including the border separator, + with the center module at (x, y). Modules can be out of bounds.""" + for dy in range(-4, 5): + for dx in range(-4, 5): + xx, yy = x + dx, y + dy + if (0 <= xx < self._size) and (0 <= yy < self._size): + # Chebyshev/infinity norm + self._set_function_module(xx, yy, max(abs(dx), abs(dy)) not in (2, 4)) + + + def _draw_alignment_pattern(self, x: int, y: int) -> None: + """Draws a 5*5 alignment pattern, with the center module + at (x, y). All modules must be in bounds.""" + for dy in range(-2, 3): + for dx in range(-2, 3): + self._set_function_module(x + dx, y + dy, max(abs(dx), abs(dy)) != 1) + + + def _set_function_module(self, x: int, y: int, isdark: bool) -> None: + """Sets the color of a module and marks it as a function module. + Only used by the constructor. Coordinates must be in bounds.""" + assert type(isdark) is bool + self._modules[y][x] = isdark + self._isfunction[y][x] = True + + + # ---- Private helper methods for constructor: Codewords and masking ---- + + def _add_ecc_and_interleave(self, data: bytearray) -> bytes: + """Returns a new byte string representing the given data with the appropriate error correction + codewords appended to it, based on this object's version and error correction level.""" + version: int = self._version + assert len(data) == QrCode._get_num_data_codewords(version, self._errcorlvl) + + # Calculate parameter numbers + numblocks: int = QrCode._NUM_ERROR_CORRECTION_BLOCKS[self._errcorlvl.ordinal][version] + blockecclen: int = QrCode._ECC_CODEWORDS_PER_BLOCK [self._errcorlvl.ordinal][version] + rawcodewords: int = QrCode._get_num_raw_data_modules(version) // 8 + numshortblocks: int = numblocks - rawcodewords % numblocks + shortblocklen: int = rawcodewords // numblocks + + # Split data into blocks and append ECC to each block + blocks: List[bytes] = [] + rsdiv: bytes = QrCode._reed_solomon_compute_divisor(blockecclen) + k: int = 0 + for i in range(numblocks): + dat: bytearray = data[k : k + shortblocklen - blockecclen + (0 if i < numshortblocks else 1)] + k += len(dat) + ecc: bytes = QrCode._reed_solomon_compute_remainder(dat, rsdiv) + if i < numshortblocks: + dat.append(0) + blocks.append(dat + ecc) + assert k == len(data) + + # Interleave (not concatenate) the bytes from every block into a single sequence + result = bytearray() + for i in range(len(blocks[0])): + for (j, blk) in enumerate(blocks): + # Skip the padding byte in short blocks + if (i != shortblocklen - blockecclen) or (j >= numshortblocks): + result.append(blk[i]) + assert len(result) == rawcodewords + return result + + + def _draw_codewords(self, data: bytes) -> None: + """Draws the given sequence of 8-bit codewords (data and error correction) onto the entire + data area of this QR Code. Function modules need to be marked off before this is called.""" + assert len(data) == QrCode._get_num_raw_data_modules(self._version) // 8 + + i: int = 0 # Bit index into the data + # Do the funny zigzag scan + for right in range(self._size - 1, 0, -2): # Index of right column in each column pair + if right <= 6: + right -= 1 + for vert in range(self._size): # Vertical counter + for j in range(2): + x: int = right - j # Actual x coordinate + upward: bool = (right + 1) & 2 == 0 + y: int = (self._size - 1 - vert) if upward else vert # Actual y coordinate + if (not self._isfunction[y][x]) and (i < len(data) * 8): + self._modules[y][x] = _get_bit(data[i >> 3], 7 - (i & 7)) + i += 1 + # If this QR Code has any remainder bits (0 to 7), they were assigned as + # 0/false/light by the constructor and are left unchanged by this method + assert i == len(data) * 8 + + + def _apply_mask(self, mask: int) -> None: + """XORs the codeword modules in this QR Code with the given mask pattern. + The function modules must be marked and the codeword bits must be drawn + before masking. Due to the arithmetic of XOR, calling _apply_mask() with + the same mask value a second time will undo the mask. A final well-formed + QR Code needs exactly one (not zero, two, etc.) mask applied.""" + if not (0 <= mask <= 7): + raise ValueError("Mask value out of range") + masker: Callable[[int,int],int] = QrCode._MASK_PATTERNS[mask] + for y in range(self._size): + for x in range(self._size): + self._modules[y][x] ^= (masker(x, y) == 0) and (not self._isfunction[y][x]) + + + def _get_penalty_score(self) -> int: + """Calculates and returns the penalty score based on state of this QR Code's current modules. + This is used by the automatic mask choice algorithm to find the mask pattern that yields the lowest score.""" + result: int = 0 + size: int = self._size + modules: List[List[bool]] = self._modules + + # Adjacent modules in row having same color, and finder-like patterns + for y in range(size): + runcolor: bool = False + runx: int = 0 + runhistory = collections.deque([0] * 7, 7) + for x in range(size): + if modules[y][x] == runcolor: + runx += 1 + if runx == 5: + result += QrCode._PENALTY_N1 + elif runx > 5: + result += 1 + else: + self._finder_penalty_add_history(runx, runhistory) + if not runcolor: + result += self._finder_penalty_count_patterns(runhistory) * QrCode._PENALTY_N3 + runcolor = modules[y][x] + runx = 1 + result += self._finder_penalty_terminate_and_count(runcolor, runx, runhistory) * QrCode._PENALTY_N3 + # Adjacent modules in column having same color, and finder-like patterns + for x in range(size): + runcolor = False + runy = 0 + runhistory = collections.deque([0] * 7, 7) + for y in range(size): + if modules[y][x] == runcolor: + runy += 1 + if runy == 5: + result += QrCode._PENALTY_N1 + elif runy > 5: + result += 1 + else: + self._finder_penalty_add_history(runy, runhistory) + if not runcolor: + result += self._finder_penalty_count_patterns(runhistory) * QrCode._PENALTY_N3 + runcolor = modules[y][x] + runy = 1 + result += self._finder_penalty_terminate_and_count(runcolor, runy, runhistory) * QrCode._PENALTY_N3 + + # 2*2 blocks of modules having same color + for y in range(size - 1): + for x in range(size - 1): + if modules[y][x] == modules[y][x + 1] == modules[y + 1][x] == modules[y + 1][x + 1]: + result += QrCode._PENALTY_N2 + + # Balance of dark and light modules + dark: int = sum((1 if cell else 0) for row in modules for cell in row) + total: int = size**2 # Note that size is odd, so dark/total != 1/2 + # Compute the smallest integer k >= 0 such that (45-5k)% <= dark/total <= (55+5k)% + k: int = (abs(dark * 20 - total * 10) + total - 1) // total - 1 + assert 0 <= k <= 9 + result += k * QrCode._PENALTY_N4 + assert 0 <= result <= 2568888 # Non-tight upper bound based on default values of PENALTY_N1, ..., N4 + return result + + + # ---- Private helper functions ---- + + def _get_alignment_pattern_positions(self) -> List[int]: + """Returns an ascending list of positions of alignment patterns for this version number. + Each position is in the range [0,177), and are used on both the x and y axes. + This could be implemented as lookup table of 40 variable-length lists of integers.""" + ver: int = self._version + if ver == 1: + return [] + else: + numalign: int = ver // 7 + 2 + step: int = 26 if (ver == 32) else \ + (ver * 4 + numalign * 2 + 1) // (numalign * 2 - 2) * 2 + result: List[int] = [(self._size - 7 - i * step) for i in range(numalign - 1)] + [6] + return list(reversed(result)) + + + @staticmethod + def _get_num_raw_data_modules(ver: int) -> int: + """Returns the number of data bits that can be stored in a QR Code of the given version number, after + all function modules are excluded. This includes remainder bits, so it might not be a multiple of 8. + The result is in the range [208, 29648]. This could be implemented as a 40-entry lookup table.""" + if not (QrCode.MIN_VERSION <= ver <= QrCode.MAX_VERSION): + raise ValueError("Version number out of range") + result: int = (16 * ver + 128) * ver + 64 + if ver >= 2: + numalign: int = ver // 7 + 2 + result -= (25 * numalign - 10) * numalign - 55 + if ver >= 7: + result -= 36 + assert 208 <= result <= 29648 + return result + + + @staticmethod + def _get_num_data_codewords(ver: int, ecl: QrCode.Ecc) -> int: + """Returns the number of 8-bit data (i.e. not error correction) codewords contained in any + QR Code of the given version number and error correction level, with remainder bits discarded. + This stateless pure function could be implemented as a (40*4)-cell lookup table.""" + return QrCode._get_num_raw_data_modules(ver) // 8 \ + - QrCode._ECC_CODEWORDS_PER_BLOCK [ecl.ordinal][ver] \ + * QrCode._NUM_ERROR_CORRECTION_BLOCKS[ecl.ordinal][ver] + + + @staticmethod + def _reed_solomon_compute_divisor(degree: int) -> bytes: + """Returns a Reed-Solomon ECC generator polynomial for the given degree. This could be + implemented as a lookup table over all possible parameter values, instead of as an algorithm.""" + if not (1 <= degree <= 255): + raise ValueError("Degree out of range") + # Polynomial coefficients are stored from highest to lowest power, excluding the leading term which is always 1. + # For example the polynomial x^3 + 255x^2 + 8x + 93 is stored as the uint8 array [255, 8, 93]. + result = bytearray([0] * (degree - 1) + [1]) # Start off with the monomial x^0 + + # Compute the product polynomial (x - r^0) * (x - r^1) * (x - r^2) * ... * (x - r^{degree-1}), + # and drop the highest monomial term which is always 1x^degree. + # Note that r = 0x02, which is a generator element of this field GF(2^8/0x11D). + root: int = 1 + for _ in range(degree): # Unused variable i + # Multiply the current product by (x - r^i) + for j in range(degree): + result[j] = QrCode._reed_solomon_multiply(result[j], root) + if j + 1 < degree: + result[j] ^= result[j + 1] + root = QrCode._reed_solomon_multiply(root, 0x02) + return result + + + @staticmethod + def _reed_solomon_compute_remainder(data: bytes, divisor: bytes) -> bytes: + """Returns the Reed-Solomon error correction codeword for the given data and divisor polynomials.""" + result = bytearray([0] * len(divisor)) + for b in data: # Polynomial division + factor: int = b ^ result.pop(0) + result.append(0) + for (i, coef) in enumerate(divisor): + result[i] ^= QrCode._reed_solomon_multiply(coef, factor) + return result + + + @staticmethod + def _reed_solomon_multiply(x: int, y: int) -> int: + """Returns the product of the two given field elements modulo GF(2^8/0x11D). The arguments and result + are unsigned 8-bit integers. This could be implemented as a lookup table of 256*256 entries of uint8.""" + if (x >> 8 != 0) or (y >> 8 != 0): + raise ValueError("Byte out of range") + # Russian peasant multiplication + z: int = 0 + for i in reversed(range(8)): + z = (z << 1) ^ ((z >> 7) * 0x11D) + z ^= ((y >> i) & 1) * x + assert z >> 8 == 0 + return z + + + def _finder_penalty_count_patterns(self, runhistory: collections.deque) -> int: + """Can only be called immediately after a light run is added, and + returns either 0, 1, or 2. A helper function for _get_penalty_score().""" + n: int = runhistory[1] + assert n <= self._size * 3 + core: bool = n > 0 and (runhistory[2] == runhistory[4] == runhistory[5] == n) and runhistory[3] == n * 3 + return (1 if (core and runhistory[0] >= n * 4 and runhistory[6] >= n) else 0) \ + + (1 if (core and runhistory[6] >= n * 4 and runhistory[0] >= n) else 0) + + + def _finder_penalty_terminate_and_count(self, currentruncolor: bool, currentrunlength: int, runhistory: collections.deque) -> int: + """Must be called at the end of a line (row or column) of modules. A helper function for _get_penalty_score().""" + if currentruncolor: # Terminate dark run + self._finder_penalty_add_history(currentrunlength, runhistory) + currentrunlength = 0 + currentrunlength += self._size # Add light border to final run + self._finder_penalty_add_history(currentrunlength, runhistory) + return self._finder_penalty_count_patterns(runhistory) + + + def _finder_penalty_add_history(self, currentrunlength: int, runhistory: collections.deque) -> None: + if runhistory[0] == 0: + currentrunlength += self._size # Add light border to initial run + runhistory.appendleft(currentrunlength) + + + # ---- Constants and tables ---- + + MIN_VERSION: int = 1 # The minimum version number supported in the QR Code Model 2 standard + MAX_VERSION: int = 40 # The maximum version number supported in the QR Code Model 2 standard + + # For use in _get_penalty_score(), when evaluating which mask is best. + _PENALTY_N1: int = 3 + _PENALTY_N2: int = 3 + _PENALTY_N3: int = 40 + _PENALTY_N4: int = 10 + + _ECC_CODEWORDS_PER_BLOCK: Sequence[Sequence[int]] = ( + # Version: (note that index 0 is for padding, and is set to an illegal value) + # 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23, 24, 25, 26, 27, 28, 29, 30, 31, 32, 33, 34, 35, 36, 37, 38, 39, 40 Error correction level + (-1, 7, 10, 15, 20, 26, 18, 20, 24, 30, 18, 20, 24, 26, 30, 22, 24, 28, 30, 28, 28, 28, 28, 30, 30, 26, 28, 30, 30, 30, 30, 30, 30, 30, 30, 30, 30, 30, 30, 30, 30), # Low + (-1, 10, 16, 26, 18, 24, 16, 18, 22, 22, 26, 30, 22, 22, 24, 24, 28, 28, 26, 26, 26, 26, 28, 28, 28, 28, 28, 28, 28, 28, 28, 28, 28, 28, 28, 28, 28, 28, 28, 28, 28), # Medium + (-1, 13, 22, 18, 26, 18, 24, 18, 22, 20, 24, 28, 26, 24, 20, 30, 24, 28, 28, 26, 30, 28, 30, 30, 30, 30, 28, 30, 30, 30, 30, 30, 30, 30, 30, 30, 30, 30, 30, 30, 30), # Quartile + (-1, 17, 28, 22, 16, 22, 28, 26, 26, 24, 28, 24, 28, 22, 24, 24, 30, 28, 28, 26, 28, 30, 24, 30, 30, 30, 30, 30, 30, 30, 30, 30, 30, 30, 30, 30, 30, 30, 30, 30, 30)) # High + + _NUM_ERROR_CORRECTION_BLOCKS: Sequence[Sequence[int]] = ( + # Version: (note that index 0 is for padding, and is set to an illegal value) + # 0, 1, 2, 3, 4, 5, 6, 7, 8, 9,10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23, 24, 25, 26, 27, 28, 29, 30, 31, 32, 33, 34, 35, 36, 37, 38, 39, 40 Error correction level + (-1, 1, 1, 1, 1, 1, 2, 2, 2, 2, 4, 4, 4, 4, 4, 6, 6, 6, 6, 7, 8, 8, 9, 9, 10, 12, 12, 12, 13, 14, 15, 16, 17, 18, 19, 19, 20, 21, 22, 24, 25), # Low + (-1, 1, 1, 1, 2, 2, 4, 4, 4, 5, 5, 5, 8, 9, 9, 10, 10, 11, 13, 14, 16, 17, 17, 18, 20, 21, 23, 25, 26, 28, 29, 31, 33, 35, 37, 38, 40, 43, 45, 47, 49), # Medium + (-1, 1, 1, 2, 2, 4, 4, 6, 6, 8, 8, 8, 10, 12, 16, 12, 17, 16, 18, 21, 20, 23, 23, 25, 27, 29, 34, 34, 35, 38, 40, 43, 45, 48, 51, 53, 56, 59, 62, 65, 68), # Quartile + (-1, 1, 1, 2, 4, 4, 4, 5, 6, 8, 8, 11, 11, 16, 16, 18, 16, 19, 21, 25, 25, 25, 34, 30, 32, 35, 37, 40, 42, 45, 48, 51, 54, 57, 60, 63, 66, 70, 74, 77, 81)) # High + + _MASK_PATTERNS: Sequence[Callable[[int,int],int]] = ( + (lambda x, y: (x + y) % 2 ), + (lambda x, y: y % 2 ), + (lambda x, y: x % 3 ), + (lambda x, y: (x + y) % 3 ), + (lambda x, y: (x // 3 + y // 2) % 2 ), + (lambda x, y: x * y % 2 + x * y % 3 ), + (lambda x, y: (x * y % 2 + x * y % 3) % 2 ), + (lambda x, y: ((x + y) % 2 + x * y % 3) % 2), + ) + + + # ---- Public helper enumeration ---- + + class Ecc: + ordinal: int # (Public) In the range 0 to 3 (unsigned 2-bit integer) + formatbits: int # (Package-private) In the range 0 to 3 (unsigned 2-bit integer) + + """The error correction level in a QR Code symbol. Immutable.""" + # Private constructor + def __init__(self, i: int, fb: int) -> None: + self.ordinal = i + self.formatbits = fb + + # Placeholders + LOW : QrCode.Ecc + MEDIUM : QrCode.Ecc + QUARTILE: QrCode.Ecc + HIGH : QrCode.Ecc + + # Public constants. Create them outside the class. + Ecc.LOW = Ecc(0, 1) # The QR Code can tolerate about 7% erroneous codewords + Ecc.MEDIUM = Ecc(1, 0) # The QR Code can tolerate about 15% erroneous codewords + Ecc.QUARTILE = Ecc(2, 3) # The QR Code can tolerate about 25% erroneous codewords + Ecc.HIGH = Ecc(3, 2) # The QR Code can tolerate about 30% erroneous codewords + + + +# ---- Data segment class ---- + +class QrSegment: + """A segment of character/binary/control data in a QR Code symbol. + Instances of this class are immutable. + The mid-level way to create a segment is to take the payload data + and call a static factory function such as QrSegment.make_numeric(). + The low-level way to create a segment is to custom-make the bit buffer + and call the QrSegment() constructor with appropriate values. + This segment class imposes no length restrictions, but QR Codes have restrictions. + Even in the most favorable conditions, a QR Code can only hold 7089 characters of data. + Any segment longer than this is meaningless for the purpose of generating QR Codes.""" + + # ---- Static factory functions (mid level) ---- + + @staticmethod + def make_bytes(data: Union[bytes,Sequence[int]]) -> QrSegment: + """Returns a segment representing the given binary data encoded in byte mode. + All input byte lists are acceptable. Any text string can be converted to + UTF-8 bytes (s.encode("UTF-8")) and encoded as a byte mode segment.""" + bb = _BitBuffer() + for b in data: + bb.append_bits(b, 8) + return QrSegment(QrSegment.Mode.BYTE, len(data), bb) + + + @staticmethod + def make_numeric(digits: str) -> QrSegment: + """Returns a segment representing the given string of decimal digits encoded in numeric mode.""" + if not QrSegment.is_numeric(digits): + raise ValueError("String contains non-numeric characters") + bb = _BitBuffer() + i: int = 0 + while i < len(digits): # Consume up to 3 digits per iteration + n: int = min(len(digits) - i, 3) + bb.append_bits(int(digits[i : i + n]), n * 3 + 1) + i += n + return QrSegment(QrSegment.Mode.NUMERIC, len(digits), bb) + + + @staticmethod + def make_alphanumeric(text: str) -> QrSegment: + """Returns a segment representing the given text string encoded in alphanumeric mode. + The characters allowed are: 0 to 9, A to Z (uppercase only), space, + dollar, percent, asterisk, plus, hyphen, period, slash, colon.""" + if not QrSegment.is_alphanumeric(text): + raise ValueError("String contains unencodable characters in alphanumeric mode") + bb = _BitBuffer() + for i in range(0, len(text) - 1, 2): # Process groups of 2 + temp: int = QrSegment._ALPHANUMERIC_ENCODING_TABLE[text[i]] * 45 + temp += QrSegment._ALPHANUMERIC_ENCODING_TABLE[text[i + 1]] + bb.append_bits(temp, 11) + if len(text) % 2 > 0: # 1 character remaining + bb.append_bits(QrSegment._ALPHANUMERIC_ENCODING_TABLE[text[-1]], 6) + return QrSegment(QrSegment.Mode.ALPHANUMERIC, len(text), bb) + + + @staticmethod + def make_segments(text: str) -> List[QrSegment]: + """Returns a new mutable list of zero or more segments to represent the given Unicode text string. + The result may use various segment modes and switch modes to optimize the length of the bit stream.""" + + # Select the most efficient segment encoding automatically + if text == "": + return [] + elif QrSegment.is_numeric(text): + return [QrSegment.make_numeric(text)] + elif QrSegment.is_alphanumeric(text): + return [QrSegment.make_alphanumeric(text)] + else: + return [QrSegment.make_bytes(text.encode("UTF-8"))] + + + @staticmethod + def make_eci(assignval: int) -> QrSegment: + """Returns a segment representing an Extended Channel Interpretation + (ECI) designator with the given assignment value.""" + bb = _BitBuffer() + if assignval < 0: + raise ValueError("ECI assignment value out of range") + elif assignval < (1 << 7): + bb.append_bits(assignval, 8) + elif assignval < (1 << 14): + bb.append_bits(0b10, 2) + bb.append_bits(assignval, 14) + elif assignval < 1000000: + bb.append_bits(0b110, 3) + bb.append_bits(assignval, 21) + else: + raise ValueError("ECI assignment value out of range") + return QrSegment(QrSegment.Mode.ECI, 0, bb) + + + # Tests whether the given string can be encoded as a segment in numeric mode. + # A string is encodable iff each character is in the range 0 to 9. + @staticmethod + def is_numeric(text: str) -> bool: + return QrSegment._NUMERIC_REGEX.fullmatch(text) is not None + + + # Tests whether the given string can be encoded as a segment in alphanumeric mode. + # A string is encodable iff each character is in the following set: 0 to 9, A to Z + # (uppercase only), space, dollar, percent, asterisk, plus, hyphen, period, slash, colon. + @staticmethod + def is_alphanumeric(text: str) -> bool: + return QrSegment._ALPHANUMERIC_REGEX.fullmatch(text) is not None + + + # ---- Private fields ---- + + # The mode indicator of this segment. Accessed through get_mode(). + _mode: QrSegment.Mode + + # The length of this segment's unencoded data. Measured in characters for + # numeric/alphanumeric/kanji mode, bytes for byte mode, and 0 for ECI mode. + # Always zero or positive. Not the same as the data's bit length. + # Accessed through get_num_chars(). + _numchars: int + + # The data bits of this segment. Accessed through get_data(). + _bitdata: List[int] + + + # ---- Constructor (low level) ---- + + def __init__(self, mode: QrSegment.Mode, numch: int, bitdata: Sequence[int]) -> None: + """Creates a new QR Code segment with the given attributes and data. + The character count (numch) must agree with the mode and the bit buffer length, + but the constraint isn't checked. The given bit buffer is cloned and stored.""" + if numch < 0: + raise ValueError() + self._mode = mode + self._numchars = numch + self._bitdata = list(bitdata) # Make defensive copy + + + # ---- Accessor methods ---- + + def get_mode(self) -> QrSegment.Mode: + """Returns the mode field of this segment.""" + return self._mode + + def get_num_chars(self) -> int: + """Returns the character count field of this segment.""" + return self._numchars + + def get_data(self) -> List[int]: + """Returns a new copy of the data bits of this segment.""" + return list(self._bitdata) # Make defensive copy + + + # Package-private function + @staticmethod + def get_total_bits(segs: Sequence[QrSegment], version: int) -> Optional[int]: + """Calculates the number of bits needed to encode the given segments at + the given version. Returns a non-negative number if successful. Otherwise + returns None if a segment has too many characters to fit its length field.""" + result = 0 + for seg in segs: + ccbits: int = seg.get_mode().num_char_count_bits(version) + if seg.get_num_chars() >= (1 << ccbits): + return None # The segment's length doesn't fit the field's bit width + result += 4 + ccbits + len(seg._bitdata) + return result + + + # ---- Constants ---- + + # Describes precisely all strings that are encodable in numeric mode. + _NUMERIC_REGEX: re.Pattern = re.compile(r"[0-9]*") + + # Describes precisely all strings that are encodable in alphanumeric mode. + _ALPHANUMERIC_REGEX: re.Pattern = re.compile(r"[A-Z0-9 $%*+./:-]*") + + # Dictionary of "0"->0, "A"->10, "$"->37, etc. + _ALPHANUMERIC_ENCODING_TABLE: Dict[str,int] = {ch: i for (i, ch) in enumerate("0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZ $%*+-./:")} + + + # ---- Public helper enumeration ---- + + class Mode: + """Describes how a segment's data bits are interpreted. Immutable.""" + + _modebits: int # The mode indicator bits, which is a uint4 value (range 0 to 15) + _charcounts: Tuple[int,int,int] # Number of character count bits for three different version ranges + + # Private constructor + def __init__(self, modebits: int, charcounts: Tuple[int,int,int]): + self._modebits = modebits + self._charcounts = charcounts + + # Package-private method + def get_mode_bits(self) -> int: + """Returns an unsigned 4-bit integer value (range 0 to 15) representing the mode indicator bits for this mode object.""" + return self._modebits + + # Package-private method + def num_char_count_bits(self, ver: int) -> int: + """Returns the bit width of the character count field for a segment in this mode + in a QR Code at the given version number. The result is in the range [0, 16].""" + return self._charcounts[(ver + 7) // 17] + + # Placeholders + NUMERIC : QrSegment.Mode + ALPHANUMERIC: QrSegment.Mode + BYTE : QrSegment.Mode + KANJI : QrSegment.Mode + ECI : QrSegment.Mode + + # Public constants. Create them outside the class. + Mode.NUMERIC = Mode(0x1, (10, 12, 14)) + Mode.ALPHANUMERIC = Mode(0x2, ( 9, 11, 13)) + Mode.BYTE = Mode(0x4, ( 8, 16, 16)) + Mode.KANJI = Mode(0x8, ( 8, 10, 12)) + Mode.ECI = Mode(0x7, ( 0, 0, 0)) + + + +# ---- Private helper class ---- + +class _BitBuffer(list): + """An appendable sequence of bits (0s and 1s). Mainly used by QrSegment.""" + + def append_bits(self, val: int, n: int) -> None: + """Appends the given number of low-order bits of the given + value to this buffer. Requires n >= 0 and 0 <= val < 2^n.""" + if (n < 0) or (val >> n != 0): + raise ValueError("Value out of range") + self.extend(((val >> i) & 1) for i in reversed(range(n))) + + +def _get_bit(x: int, i: int) -> bool: + """Returns true iff the i'th bit of x is set to 1.""" + return (x >> i) & 1 != 0 + + + +class DataTooLongError(ValueError): + """Raised when the supplied data does not fit any QR Code version. Ways to handle this exception include: + - Decrease the error correction level if it was greater than Ecc.LOW. + - If the encode_segments() function was called with a maxversion argument, then increase + it if it was less than QrCode.MAX_VERSION. (This advice does not apply to the other + factory functions because they search all versions up to QrCode.MAX_VERSION.) + - Split the text data into better or optimal segments in order to reduce the number of bits required. + - Change the text or binary data to be shorter. + - Change the text to fit the character set of a particular segment mode (e.g. alphanumeric). + - Propagate the error upward to the caller/user.""" + pass diff --git a/resources/lib/twitch_addon/routes/device_login.py b/resources/lib/twitch_addon/routes/device_login.py index 536448b4..f161623c 100644 --- a/resources/lib/twitch_addon/routes/device_login.py +++ b/resources/lib/twitch_addon/routes/device_login.py @@ -9,17 +9,36 @@ See LICENSES/GPL-3.0-only for more information. """ +import os +import time +import threading + import xbmc import xbmcgui +import xbmcvfs from ..addon import utils, device_oauth from ..addon.common import kodi, log_utils -from ..addon.constants import SCOPES +from ..addon.constants import SCOPES, ADDON_DATA_DIR from ..addon.utils import i18n +try: + from ..addon import qr as qr_render +except Exception: + qr_render = None + +# Short, clean activation URL (Twitch's verification_uri carries the code as a long query string). +_ACTIVATE_URL = 'https://www.twitch.tv/activate' + +_ALIGN_CENTER_X = 0x00000002 # XBFONT_CENTER_X +ACTION_PREVIOUS_MENU = 10 +ACTION_STOP = 13 +ACTION_NAV_BACK = 92 + def do_device_login(heading, body_i18n, client_id, scopes, store_tokens, success_i18n, log_tag): - """Shared device-code login flow: show code, poll, store tokens via store_tokens().""" + """Shared device-code login flow: show a QR code (with a text-dialog fallback), + poll for the token, and store it via store_tokens().""" ok, data = device_oauth.request_device_code(client_id, scopes) if not ok or not data.get('user_code'): kodi.Dialog().ok(heading, i18n('device_login_unsupported')) @@ -29,53 +48,221 @@ def do_device_login(heading, body_i18n, client_id, scopes, store_tokens, success device_code = data['device_code'] interval = int(data.get('interval', 5)) or 5 expires_in = int(data.get('expires_in', 1800)) or 1800 + verification_uri = data.get('verification_uri') or _ACTIVATE_URL + body = i18n(body_i18n) % (_ACTIVATE_URL, user_code) + + poll = _make_poller(client_id, scopes, device_code) + + # Prefer the QR window; fall back to the plain text dialog when QR can't be + # rendered/shown. A QR user-cancel or expiry is a definitive end (no fallback). + result, token = 'fallback', None + qr_path = _render_qr(verification_uri) + if qr_path: + try: + result, token = _qr_login(heading, qr_path, verification_uri, user_code, + expires_in, interval, poll) + except Exception as err: + log_utils.log('QR login window failed, using text fallback: %s' % repr(err), + log_utils.LOGWARNING) + result = 'fallback' + finally: + _remove(qr_path) + if result == 'fallback': + result, token = _text_login(heading, body, expires_in, interval, poll) + + if result == 'ok' and token and token.get('access_token'): + store_tokens(token['access_token'], + token.get('refresh_token', ''), + token.get('expires_in', 14400)) + log_utils.log('OAuth%s: device login succeeded, refresh token stored' % log_tag, + log_utils.LOGNOTICE) + kodi.Dialog().ok(heading, i18n(success_i18n)) + elif result == 'cancelled': + kodi.notify(i18n('login'), i18n('device_login_cancelled'), sound=False) + else: + kodi.notify(i18n('login'), i18n('device_login_failed'), sound=False) - # Short, clean activation URL (Twitch's verification_uri carries the code as a long query string). - activate_url = 'https://www.twitch.tv/activate' - body = i18n(body_i18n) % (activate_url, user_code) +def _make_poller(client_id, scopes, device_code): + """Return poll() -> (state, token) where state is 'ok' | 'slow_down' | 'stop' | 'pending'.""" + def poll(): + status, tdata = device_oauth.poll_device_token(client_id, scopes, device_code) + if status == 'ok': + return 'ok', tdata + if status == 'slow_down': + return 'slow_down', None + if status in ('expired', 'denied'): + return 'stop', None + return 'pending', None + return poll + + +def _text_login(heading, body, expires_in, interval, poll): + """Modal progress-dialog fallback that works on every skin/device (and when a + camera isn't handy -- the viewer just types the short code). Returns (result, token).""" monitor = xbmc.Monitor() progress = xbmcgui.DialogProgress() progress.create(heading, body) - - token = None waited = 0 while waited < expires_in: - # wait the poll interval, abortable (user cancel or Kodi shutdown) for _ in range(interval): if progress.iscanceled(): progress.close() - kodi.notify(i18n('login'), i18n('device_login_cancelled'), sound=False) - return + return 'cancelled', None if monitor.waitForAbort(1): # Kodi is shutting down progress.close() - return + return 'failed', None waited += 1 - status, tdata = device_oauth.poll_device_token(client_id, scopes, device_code) - if status == 'ok': - token = tdata - break - elif status == 'slow_down': + state, token = poll() + if state == 'ok': + progress.close() + return 'ok', token + elif state == 'slow_down': interval += 2 - elif status in ('expired', 'denied'): + elif state == 'stop': break - # 'pending' / transient 'error' -> keep polling try: progress.update(int(min(99, (waited * 100) // expires_in)), body + '[CR]' + i18n('device_login_waiting')) except Exception: pass - progress.close() - if token and token.get('access_token'): - store_tokens(token['access_token'], - token.get('refresh_token', ''), - token.get('expires_in', 14400)) - log_utils.log('OAuth%s: device login succeeded, refresh token stored' % log_tag, - log_utils.LOGNOTICE) - kodi.Dialog().ok(heading, i18n(success_i18n)) - else: - kodi.notify(i18n('login'), i18n('device_login_failed'), sound=False) + return 'failed', None + + +def _render_qr(verification_uri): + if qr_render is None: + return None + try: + if not xbmcvfs.exists(ADDON_DATA_DIR): + xbmcvfs.mkdirs(ADDON_DATA_DIR) + qr_path = os.path.join(ADDON_DATA_DIR, 'device_login_qr.png') + return qr_render.generate_qr_png(verification_uri, qr_path, scale=8, border=4) + except Exception as err: + log_utils.log('QR render failed: %s' % repr(err), log_utils.LOGWARNING) + return None + + +def _remove(path): + try: + if path and os.path.exists(path): + os.remove(path) + except Exception: + pass + + +def _qr_login(heading, qr_path, verification_uri, user_code, expires_in, interval, poll): + dialog = _QRLoginDialog() + try: + dialog.setup(heading, qr_path, verification_uri, user_code, expires_in, interval, poll) + dialog.start_polling() + dialog.doModal() + dialog.stop = True + return dialog.result, dialog.token + finally: + _remove(dialog.bg_path) + del dialog + + +class _QRLoginDialog(xbmcgui.WindowDialog): + def __init__(self): + super(_QRLoginDialog, self).__init__() + self.result = 'failed' + self.token = None + self.stop = False + self.bg_path = None + self._poll = None + self._interval = 5 + self._expires_in = 1800 + self._thread = None + + def setup(self, heading, qr_path, verification_uri, user_code, expires_in, interval, poll): + self._poll = poll + self._interval = interval + self._expires_in = expires_in + + width = self.getWidth() + height = self.getHeight() + center_x = width // 2 + + panel_w = min(760, width - 40) + panel_h = min(640, height - 40) + panel_x = center_x - panel_w // 2 + panel_y = (height - panel_h) // 2 + + qr_size = min(360, panel_h - 260) + text_x = panel_x + 30 + text_w = panel_w - 60 + + # Solid backdrop panel rendered to its own PNG so we don't rely on skin textures. + if qr_render is not None: + try: + self.bg_path = os.path.join(ADDON_DATA_DIR, 'device_login_bg.png') + qr_render.generate_solid_png(self.bg_path, 8, 8, value=20) + self.addControl(xbmcgui.ControlImage(panel_x, panel_y, panel_w, panel_h, + self.bg_path, aspectRatio=0)) + except Exception: + self.bg_path = None + + self.addControl(xbmcgui.ControlLabel( + text_x, panel_y + 22, text_w, 40, heading, + alignment=_ALIGN_CENTER_X, textColor='FFFFFFFF')) + + qr_x = center_x - qr_size // 2 + qr_y = panel_y + 74 + self.addControl(xbmcgui.ControlImage(qr_x, qr_y, qr_size, qr_size, qr_path, aspectRatio=0)) + + info_y = qr_y + qr_size + 14 + self.addControl(xbmcgui.ControlLabel( + text_x, info_y, text_w, 30, i18n('device_login_scan') % verification_uri, + alignment=_ALIGN_CENTER_X, textColor='FFDDDDDD')) + self.addControl(xbmcgui.ControlLabel( + text_x, info_y + 36, text_w, 44, i18n('device_login_code') % user_code, + alignment=_ALIGN_CENTER_X, textColor='FF9147FF')) + self.addControl(xbmcgui.ControlLabel( + text_x, info_y + 86, text_w, 30, i18n('device_login_waiting_cancel'), + alignment=_ALIGN_CENTER_X, textColor='FFAAAAAA')) + + def start_polling(self): + self._thread = threading.Thread(target=self._run) + self._thread.daemon = True + self._thread.start() + + def _run(self): + # Let doModal() take over before this thread is allowed to close the window. + xbmc.sleep(400) + start = time.time() + while not self.stop: + state, token = self._safe_poll() + if state == 'ok': + self.token = token + self.result = 'ok' + break + if state == 'stop': + self.result = 'failed' + break + if state == 'slow_down': + self._interval += 2 + if time.time() - start >= self._expires_in: + self.result = 'failed' + break + waited = 0.0 + while waited < self._interval and not self.stop: + xbmc.sleep(500) + waited += 0.5 + self.close() + + def _safe_poll(self): + try: + return self._poll() + except Exception: + return 'pending', None + + def onAction(self, action): + if action.getId() in (ACTION_PREVIOUS_MENU, ACTION_NAV_BACK, ACTION_STOP): + self.result = 'cancelled' + self.stop = True + self.close() def route():