kuhyx/testsAndMisc
1
0
Fork 0
mirror of https://github.com/kuhyx/testsAndMisc.git synced 2026-11-10 19:27:03 +01:00
Code Issues Actions 6 Packages Projects Releases Wiki Activity

Make focus policy and blocklist usable without root; fix workout CDN unblock

Browse Source 
phone_focus_mode is built for a rooted phone: the hosts layer is a read-only
bind mount over /system/etc/hosts, chosen so that even root-in-a-terminal
cannot overwrite it. None of that ports to a stock locked device. This splits
out the parts that are device-independent so one policy can drive both.

python_pkg/focus_policy parses config.sh rather than copying it, so the shell
config stays the source of truth and the two cannot drift. It rejects the
shipped REDACTED_* coordinate placeholder instead of defaulting to (0,0),
which would move "home" to the Atlantic and quietly disable the location gate.

blocklist.py converts the hosts file to domains for a DoT resolver or a local
VpnService. Three properties of the real 194k-line file make the obvious parser
wrong, and all three are handled: generate_hosts_file.sh *comments out*
deliberate unblocks rather than deleting them, so stripping '#' re-blocks
facebook/messenger/linkedin/4chan; the StevenBlack header carries loopback and
IPv6 housekeeping that would blackhole localhost; and ~1060 upstream annotation
tokens ('#[adware.zeno]', '"spam') render as address=/"spam/, which makes
dnsmasq refuse to start -- taking DNS down rather than blocking anything.
Verified with dnsmasq --test over all 185681 generated domains.

strip_workout_hosts.py had the same class of bug on the live rooted phone:
exact set membership meant the allowlisted googlevideo.com never matched the
dynamic r*---sn-*.googlevideo.com hosts YouTube actually streams from, so the
workout unblock was only partly working. Now suffix-matched (17 lines stripped,
was 14). The three new tests were checked against the original code in a
throwaway worktree; two of them fail there, so they pin the fix rather than
passing vacuously.

focus_owner is a device-owner app that deliberately does nothing yet. It ships
clearDeviceOwnerApp() implemented and tested first, because dpm
remove-active-admin does not work on a device owner: an app that locks the
device down before proving it can let go leaves a soft-brick recoverable only
through fastboot on a locked bootloader. Built, installed and launched on the
Pixel 6a; dumpsys afterwards confirms Device Owner Type -1 and no user
restrictions, i.e. the phone is unchanged.

Measured while scoping this, and not yet reflected in any enforcement code:
on Android 16 both pm suspend and pm disable-user are cleared by a reboot, so
shell-level app blocking is per-session only. setApplicationHidden is assumed
to persist under Device Owner but is untested.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01AwepyL5Lxs5QKbKkMpxxxL
This commit is contained in:
Krzysztof Rudnicki 2026-08-05 09:15:02 +02:00
parent f1590cbb64
commit f636d2b21b
39 changed files with 2744 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

24
docs/superpowers/contracts/unrooted-focus-mode-policy-and-owner-app-2026-08-05.json Normal file
View File

@ -0,0 +1,24 @@
{
"title": "Device-independent focus policy, blocklist pipeline, and inert device-owner app",
"objective": "Extract the parts of phone_focus_mode that do not require root into a shared, tested policy layer, so that one definition can drive both the rooted shell enforcer and a future unrooted Device Owner backend. Deliver the blocklist as domains rather than a hosts file, since an unrooted phone cannot bind-mount /system/etc/hosts. Ship a device-owner app whose removal path is implemented and verified before any restriction exists. Success is: the policy loads from the real config.sh, the generated blocklist is accepted by the resolver that will consume it, and the app runs on the Pixel 6a while leaving the device provably unmodified.",
"acceptance_criteria": [
"python_pkg/focus_policy parses the real phone_focus_mode/config.sh - it does not duplicate the package lists, so the shell config remains the single source of truth",
"The shipped REDACTED_* coordinate placeholder raises a PolicyError rather than defaulting to (0,0), which would silently disable the location gate",
"Commented-out hosts entries are honoured as deliberate unblocks, so facebook/messenger/linkedin/4chan stay unblocked after conversion",
"Loopback and IPv6 housekeeping lines never become blocked domains",
"Invalid upstream tokens are rejected, and dnsmasq --test accepts the generated config over the full real blocklist",
"The workout exception releases subdomains, so dynamic r*---sn-*.googlevideo.com CDN hosts are unblocked during a workout",
"The same suffix-matching fix is applied to the live phone_focus_mode/strip_workout_hosts.py, with regression tests that fail against the pre-fix code",
"100% branch coverage on python_pkg/focus_policy with zero lint suppressions",
"focus_owner implements clearDeviceOwnerApp() and it is unit-tested before any restriction-setting code exists",
"focus_owner builds, installs, and runs on the Pixel 6a, and dumpsys afterwards shows no device owner, no user restrictions, and no suspended packages"
],
"out_of_scope": [
"Device Owner provisioning - dpm set-device-owner requires a factory reset that has not been agreed",
"Any enforcement whatsoever: no setApplicationHidden, no addUserRestriction, no setAlwaysOnVpnPackage",
"Changes to the rooted enforcement layers - hosts bind mount, iptables chains, netd restart are untouched",
"The DoT endpoint and the dns_enforcer port 853 conflict, both tracked separately",
"Shell-side fixes to generate_hosts_file.sh - only strip_workout_hosts.py is changed here"
],
"verifier": "pre-commit run --files <staged>; coverage run --branch --source=python_pkg/focus_policy -m pytest python_pkg/focus_policy/tests/ && coverage report (expect 100%); pytest linux_configuration/tests/test_strip_workout_hosts.py; dnsmasq --test --conf-file=<generated>; cd focus_owner && flutter analyze && flutter test; adb install + am start + dumpsys package/device_policy to confirm the device is unchanged"
}

60
docs/superpowers/evidence/unrooted-focus-mode-policy-and-owner-app-2026-08-05.json Normal file
View File

@ -0,0 +1,60 @@
{
"intent": "Make focus mode's policy and blocklist reusable on an unrooted phone, and ship a device-owner app whose removal path is proven before any restriction exists. Also fix a live bug that left YouTube's CDN blocked during workouts on the rooted phone.",
"scope": [
"python_pkg/focus_policy/ - new: typed policy model, config.sh loader, hosts->domain blocklist, JSON export",
"focus_owner/ - new: com.kuhy.focus_owner Flutter+Kotlin app, INERT (no restrictions applied, not provisioned)",
"phone_focus_mode/strip_workout_hosts.py - fix exact-match domain filter",
"linux_configuration/tests/test_strip_workout_hosts.py - regression tests for the above",
"Non-goal: no Device Owner provisioning, no factory reset, no enforcement of any kind",
"Non-goal: shell-side hosts/iptables enforcement is unchanged"
],
"changes": [
"focus_policy parses phone_focus_mode/config.sh rather than duplicating its lists, so the shell config stays the single source of truth. HOME_LAT/LON load from the gitignored config_secrets.sh; the shipped REDACTED_* placeholder is rejected loudly instead of defaulting to (0,0), which would silently relocate 'home' to the Atlantic and disable the location gate.",
"blocklist.py converts the hosts file to domains. It honours commented-out entries as deliberate unblocks (generate_hosts_file.sh comments rather than deletes, so a naive parser re-blocks facebook/linkedin/4chan), skips loopback and IPv6 housekeeping lines, and rejects ~1060 upstream annotation tokens such as '#[adware.zeno]' and '\"spam' that would otherwise render as address=/\"spam/ and make dnsmasq refuse to start.",
"strip_workout_hosts.py now matches domain suffixes instead of exact set membership. YouTube streams from dynamic r*---sn-*.googlevideo.com hosts, which never equalled the allowlisted 'googlevideo.com', so the workout unblock was only partially effective on the live rooted phone.",
"focus_owner ships DevicePolicyBridge.releaseDeviceOwner() (clearDeviceOwnerApp) implemented and tested BEFORE any restriction. dpm remove-active-admin does not work on a device owner, so an app that locks down before proving it can let go leaves a soft-brick recoverable only via fastboot on a locked bootloader."
],
"verification": [
{
"command": "pre-commit run --files python_pkg/focus_policy/*.py python_pkg/focus_policy/tests/*.py",
"result": "pass",
"evidence": "All hooks pass including ruff, ruff format, mypy (strict), pylint, bandit, no-noqa. Zero lint suppressions: the three initially-added ones (S104, two type: ignore) were removed by fixing the underlying code, not by suppressing."
},
{
"command": "coverage run --branch --source=python_pkg/focus_policy -m pytest python_pkg/focus_policy/tests/ && coverage report",
"result": "pass",
"evidence": "109 passed. 219 statements, 56 branches, 0 missed = 100% branch coverage across all 5 modules."
},
{
"command": "dnsmasq --test --conf-file=<generated blocklist.conf>",
"result": "pass",
"evidence": "'dnsmasq: syntax check OK.' over all 185681 domains generated from the real 194463-line hosts file. An earlier run FAILED with 'missing \" at line 1' until invalid-domain rejection was added - this check is what caught it."
},
{
"command": "pytest linux_configuration/tests/test_strip_workout_hosts.py",
"result": "pass",
"evidence": "8 passed (5 pre-existing + 3 new). Non-vacuity confirmed: the 3 new tests were run against the ORIGINAL code in a throwaway git worktree and 2 of 3 FAILED there, so they genuinely pin the fix."
},
{
"command": "cd focus_owner && flutter analyze && flutter test",
"result": "pass",
"evidence": "'No issues found!' and '+10: All tests passed!' covering status parsing, the release path, null-reply handling, and the confirm-before-release dialog."
},
{
"command": "flutter build apk --debug && adb -s 23181JEGR08034 install -r && am start",
"result": "pass",
"evidence": "Built, installed, and launched on the Pixel 6a. Screenshot shows live platform-channel data ('Android SDK 36' read from the device), 'Device owner: no', 'Restrictions: none (inert build)'. Post-run dumpsys confirms Device Owner Type: -1, user restrictions 'none', mAppsSuspended=false - the phone is unchanged."
}
],
"risks": [
"focus_owner declares a DeviceAdminReceiver. Declaring it is inert - it applies nothing until someone runs dpm set-device-owner, which cannot succeed while the device has accounts. Still, the component now exists and should not be provisioned until the release path is exercised on a real device.",
"The suffix matching in strip_workout_hosts.py releases a broader set than before (17 lines stripped vs 14). That is the intended fix, but it means any future entry in WORKOUT_UNBLOCK_DOMAINS now releases its entire subtree - a short domain there would release a lot.",
"focus_policy duplicates the Haversine and curfew semantics of focus_daemon.sh in Python. The two must stay in agreement; the loader reads the same config.sh, but the geometry is a reimplementation, not a shared one.",
"Measured this session and NOT yet reflected in any enforcement code: both pm suspend and pm disable-user are cleared by a reboot on Android 16. setApplicationHidden is assumed to persist but is UNTESTED."
],
"rollback": [
"git revert the commit. focus_policy and focus_owner are new directories with no other module importing them, so removing them affects nothing else.",
"The only change to live behaviour is strip_workout_hosts.py, used by deploy.sh when generating the workout hosts variant on the PC. Reverting restores exact-match filtering (i.e. restores the bug).",
"After rollback validate: pytest linux_configuration/tests/test_strip_workout_hosts.py still passes, and phone_focus_mode/deploy.sh still produces a workout variant."
]
}

45
focus_owner/.gitignore vendored Normal file
View File

@ -0,0 +1,45 @@
# Miscellaneous
*.class
*.log
*.pyc
*.swp
.DS_Store
.atom/
.build/
.buildlog/
.history
.svn/
.swiftpm/
migrate_working_dir/
# IntelliJ related
*.iml
*.ipr
*.iws
.idea/
# The .vscode folder contains launch configuration and tasks you configure in
# VS Code which you may wish to be included in version control, so this line
# is commented out by default.
#.vscode/
# Flutter/Dart/Pub related
**/doc/api/
**/ios/Flutter/.last_build_id
.dart_tool/
.flutter-plugins-dependencies
.pub-cache/
.pub/
/build/
/coverage/
# Symbolication related
app.*.symbols
# Obfuscation related
app.*.map.json
# Android Studio will place build artifacts here
/android/app/debug
/android/app/profile
/android/app/release

30
focus_owner/.metadata Normal file
View File

@ -0,0 +1,30 @@
# This file tracks properties of this Flutter project.
# Used by Flutter tool to assess capabilities and perform upgrades etc.
#
# This file should be version controlled and should not be manually edited.
version:
revision: "058e0af2c2b57e369d905a03ac9748b0ebf543c6"
channel: "stable"
project_type: app
# Tracks metadata for the flutter migrate command
migration:
platforms:
- platform: root
create_revision: 058e0af2c2b57e369d905a03ac9748b0ebf543c6
base_revision: 058e0af2c2b57e369d905a03ac9748b0ebf543c6
- platform: android
create_revision: 058e0af2c2b57e369d905a03ac9748b0ebf543c6
base_revision: 058e0af2c2b57e369d905a03ac9748b0ebf543c6
# User provided section
# List of Local paths (relative to this file) that should be
# ignored by the migrate tool.
#
# Files that are not part of the templates will be ignored by default.
unmanaged_files:
- 'lib/main.dart'
- 'ios/Runner.xcodeproj/project.pbxproj'

56
focus_owner/README.md Normal file
View File

@ -0,0 +1,56 @@
# focus_owner
Device Owner companion app for unrooted focus mode. **Currently inert**: it can
be provisioned as device owner, but it applies no restrictions.
## Why the escape hatch ships first
`dpm remove-active-admin` does not work on a device owner. The only exit that
avoids a factory reset is `DevicePolicyManager.clearDeviceOwnerApp()`, callable
solely by this package on itself. An app that locks the device down before
proving it can let go leaves a soft-brick recoverable only through fastboot —
painful on a locked bootloader, and impossible to reach if
`DISALLOW_FACTORY_RESET` is ever set.
So: `DevicePolicyBridge.releaseDeviceOwner()` is implemented, unit-tested, and
reachable from the app's main screen with no PC attached. Enforcement is added
only after that path has been exercised on a provisioned device.
## Provisioning (NOT done yet — requires a factory reset)
`dpm set-device-owner` refuses to run while any account exists on the device.
The Pixel 6a has 6, so provisioning means a wipe. See
`~/.claude/projects/-home-kuhy-testsAndMisc/memory/device-owner-not-root-banking-works.md`
for what does and does not survive that, and for the Google Wallet open risk.
Once wiped, before adding any account:
```bash
adb install -r build/app/outputs/flutter-apk/app-debug.apk
adb shell dpm set-device-owner com.kuhy.focus_owner/.FocusDeviceAdminReceiver
```
Then immediately verify tap-to-pay and banking apps, while the release button
is still reachable.
## Build and verify
```bash
cd ~/testsAndMisc/focus_owner
flutter analyze && flutter test
flutter build apk --debug
adb -s 23181JEGR08034 install -r build/app/outputs/flutter-apk/app-debug.apk
adb -s 23181JEGR08034 shell am start -n com.kuhy.focus_owner/.MainActivity
```
The status screen reads live state over the platform channel; "Android SDK 36"
confirms the Kotlin bridge is working.
## Still to do
- Consume the policy JSON from `python_pkg/focus_policy` (`export.py`).
- Add enforcement (`setApplicationHidden`, `addUserRestriction`,
`setAlwaysOnVpnPackage`) — only after the release path is verified on-device.
- Verify `setApplicationHidden` survives a reboot. `pm suspend` and
`pm disable-user` both do NOT; do not assume the DPM equivalent behaves the
same until measured.

28
focus_owner/analysis_options.yaml Normal file
View File

@ -0,0 +1,28 @@
# This file configures the analyzer, which statically analyzes Dart code to
# check for errors, warnings, and lints.
#
# The issues identified by the analyzer are surfaced in the UI of Dart-enabled
# IDEs (https://dart.dev/tools#ides-and-editors). The analyzer can also be
# invoked from the command line by running `flutter analyze`.
# The following line activates a set of recommended lints for Flutter apps,
# packages, and plugins designed to encourage good coding practices.
include: package:flutter_lints/flutter.yaml
linter:
# The lint rules applied to this project can be customized in the
# section below to disable rules from the `package:flutter_lints/flutter.yaml`
# included above or to enable additional rules. A list of all available lints
# and their documentation is published at https://dart.dev/lints.
#
# Instead of disabling a lint rule for the entire project in the
# section below, it can also be suppressed for a single line of code
# or a specific dart file by using the `// ignore: name_of_lint` and
# `// ignore_for_file: name_of_lint` syntax on the line or in the file
# producing the lint.
rules:
# avoid_print: false # Uncomment to disable the `avoid_print` rule
# prefer_single_quotes: true # Uncomment to enable the `prefer_single_quotes` rule
# Additional information about this file can be found at
# https://dart.dev/guides/language/analysis-options

14
focus_owner/android/.gitignore vendored Normal file
View File

@ -0,0 +1,14 @@
gradle-wrapper.jar
/.gradle
/captures/
/gradlew
/gradlew.bat
/local.properties
GeneratedPluginRegistrant.java
.cxx/
# Remember to never publicly share your keystore.
# See https://flutter.dev/to/reference-keystore
key.properties
**/*.keystore
**/*.jks

45
focus_owner/android/app/build.gradle.kts Normal file
View File

@ -0,0 +1,45 @@
plugins {
id("com.android.application")
// The Flutter Gradle Plugin must be applied after the Android and Kotlin Gradle plugins.
id("dev.flutter.flutter-gradle-plugin")
}
android {
namespace = "com.kuhy.focus_owner"
compileSdk = flutter.compileSdkVersion
ndkVersion = flutter.ndkVersion
compileOptions {
sourceCompatibility = JavaVersion.VERSION_17
targetCompatibility = JavaVersion.VERSION_17
}
defaultConfig {
// TODO: Specify your own unique Application ID (https://developer.android.com/studio/build/application-id.html).
applicationId = "com.kuhy.focus_owner"
// You can update the following values to match your application needs.
// For more information, see: https://flutter.dev/to/review-gradle-config.
minSdk = flutter.minSdkVersion
targetSdk = flutter.targetSdkVersion
versionCode = flutter.versionCode
versionName = flutter.versionName
}
buildTypes {
release {
// TODO: Add your own signing config for the release build.
// Signing with the debug keys for now, so `flutter run --release` works.
signingConfig = signingConfigs.getByName("debug")
}
}
}
kotlin {
compilerOptions {
jvmTarget = org.jetbrains.kotlin.gradle.dsl.JvmTarget.JVM_17
}
}
flutter {
source = "../.."
}

7
focus_owner/android/app/src/debug/AndroidManifest.xml Normal file
View File

@ -0,0 +1,7 @@
<manifest xmlns:android="http://schemas.android.com/apk/res/android">
<!-- The INTERNET permission is required for development. Specifically,
the Flutter tool needs it to communicate with the running application
to allow setting breakpoints, to provide hot reload, etc.
-->
<uses-permission android:name="android.permission.INTERNET"/>
</manifest>

61
focus_owner/android/app/src/main/AndroidManifest.xml Normal file
View File

@ -0,0 +1,61 @@
<manifest xmlns:android="http://schemas.android.com/apk/res/android">
<application
android:label="focus_owner"
android:name="${applicationName}"
android:icon="@mipmap/ic_launcher">
<activity
android:name=".MainActivity"
android:exported="true"
android:launchMode="singleTop"
android:taskAffinity=""
android:theme="@style/LaunchTheme"
android:configChanges="orientation|keyboardHidden|keyboard|screenSize|smallestScreenSize|locale|layoutDirection|fontScale|screenLayout|density|uiMode"
android:hardwareAccelerated="true"
android:windowSoftInputMode="adjustResize">
<!-- Specifies an Android theme to apply to this Activity as soon as
the Android process has started. This theme is visible to the user
while the Flutter UI initializes. After that, this theme continues
to determine the Window background behind the Flutter UI. -->
<meta-data
android:name="io.flutter.embedding.android.NormalTheme"
android:resource="@style/NormalTheme"
/>
<intent-filter>
<action android:name="android.intent.action.MAIN"/>
<category android:name="android.intent.category.LAUNCHER"/>
</intent-filter>
</activity>
<!-- Device admin component. Declaring it makes device-owner
provisioning possible; it applies no restrictions on its own.
BIND_DEVICE_ADMIN must be held by the system, and the receiver must
be exported, or `dpm set-device-owner` cannot find it. -->
<receiver
android:name=".FocusDeviceAdminReceiver"
android:exported="true"
android:permission="android.permission.BIND_DEVICE_ADMIN">
<meta-data
android:name="android.app.device_admin"
android:resource="@xml/device_admin" />
<intent-filter>
<action android:name="android.app.action.DEVICE_ADMIN_ENABLED" />
</intent-filter>
</receiver>
<!-- Don't delete the meta-data below.
This is used by the Flutter tool to generate GeneratedPluginRegistrant.java -->
<meta-data
android:name="flutterEmbedding"
android:value="2" />
</application>
<!-- Required to query activities that can process text, see:
https://developer.android.com/training/package-visibility and
https://developer.android.com/reference/android/content/Intent#ACTION_PROCESS_TEXT.
In particular, this is used by the Flutter engine in io.flutter.plugin.text.ProcessTextPlugin. -->
<queries>
<intent>
<action android:name="android.intent.action.PROCESS_TEXT"/>
<data android:mimeType="text/plain"/>
</intent>
</queries>
</manifest>

74
focus_owner/android/app/src/main/kotlin/com/kuhy/focus_owner/DevicePolicyBridge.kt Normal file
View File

@ -0,0 +1,74 @@
package com.kuhy.focus_owner
import android.app.admin.DevicePolicyManager
import android.content.ComponentName
import android.content.Context
import android.os.Build
import android.util.Log
/**
* Thin wrapper over [DevicePolicyManager] exposing only what this build needs:
* reporting provisioning state, and giving up device ownership.
*
* No restriction is applied anywhere in this class. Adding enforcement is a
* later, separate step; see [FocusDeviceAdminReceiver] for why the release path
* ships and is verified first.
*/
class DevicePolicyBridge(private val context: Context) {
private val dpm: DevicePolicyManager =
context.getSystemService(Context.DEVICE_POLICY_SERVICE) as DevicePolicyManager
private val admin: ComponentName =
ComponentName(context, FocusDeviceAdminReceiver::class.java)
/** Whether this package currently holds device owner. */
fun isDeviceOwner(): Boolean = dpm.isDeviceOwnerApp(context.packageName)
/** Whether the admin component is active (a weaker state than ownership). */
fun isAdminActive(): Boolean = dpm.isAdminActive(admin)
/**
* Describes provisioning state for the UI, so the escape hatch can be found
* without a PC attached.
*/
fun status(): Map<String, Any> = mapOf(
"packageName" to context.packageName,
"isDeviceOwner" to isDeviceOwner(),
"isAdminActive" to isAdminActive(),
"sdkInt" to Build.VERSION.SDK_INT,
"restrictionsApplied" to false,
)
/**
* Relinquishes device ownership without wiping the device.
*
* This is the only exit that preserves user data: `dpm remove-active-admin`
* refuses to act on a device owner, and a factory reset is the sole
* alternative — one that `DISALLOW_FACTORY_RESET` could itself remove.
*
* [DevicePolicyManager.clearDeviceOwnerApp] is deprecated (API 26) but
* still functional, and it is callable only by the owning package on
* itself, which is exactly why this method has to live inside the app.
*
* @return true when ownership was released or was never held.
*/
@Suppress("DEPRECATION")
fun releaseDeviceOwner(): Boolean {
if (!isDeviceOwner()) {
Log.i(FocusDeviceAdminReceiver.TAG, "Not device owner; nothing to release")
return true
}
return runCatching {
dpm.clearDeviceOwnerApp(context.packageName)
val released = !isDeviceOwner()
Log.i(FocusDeviceAdminReceiver.TAG, "clearDeviceOwnerApp -> released=$released")
released
}.getOrElse { error ->
// Never rethrow: the UI button that calls this is the last resort
// before a factory reset, so it must always render an outcome.
Log.e(FocusDeviceAdminReceiver.TAG, "clearDeviceOwnerApp failed", error)
false
}
}
}

37
focus_owner/android/app/src/main/kotlin/com/kuhy/focus_owner/FocusDeviceAdminReceiver.kt Normal file
View File

@ -0,0 +1,37 @@
package com.kuhy.focus_owner
import android.app.admin.DeviceAdminReceiver
import android.content.Context
import android.content.Intent
import android.util.Log
/**
* Device admin component for focus mode.
*
* Deliberately inert: it declares the component so the app *can* be provisioned
* as device owner, but it applies no user restrictions and hides no packages.
* Enforcement is added only after the removal path in [DevicePolicyBridge] has
* been exercised on a real device.
*
* The ordering matters and is not stylistic. `dpm remove-active-admin` does not
* work on a device owner, so the only exit that avoids a factory reset is
* `clearDeviceOwnerApp()`, callable solely by this package on itself. An app
* that locks the device down before proving it can let go leaves a soft-brick
* recoverable only through fastboot — painful on a locked bootloader.
*/
class FocusDeviceAdminReceiver : DeviceAdminReceiver() {
override fun onEnabled(context: Context, intent: Intent) {
super.onEnabled(context, intent)
Log.i(TAG, "Device admin enabled")
}
override fun onDisabled(context: Context, intent: Intent) {
super.onDisabled(context, intent)
Log.i(TAG, "Device admin disabled")
}
companion object {
const val TAG = "FocusOwner"
}
}

25
focus_owner/android/app/src/main/kotlin/com/kuhy/focus_owner/MainActivity.kt Normal file
View File

@ -0,0 +1,25 @@
package com.kuhy.focus_owner
import io.flutter.embedding.android.FlutterActivity
import io.flutter.embedding.engine.FlutterEngine
import io.flutter.plugin.common.MethodChannel
class MainActivity : FlutterActivity() {
override fun configureFlutterEngine(flutterEngine: FlutterEngine) {
super.configureFlutterEngine(flutterEngine)
val bridge = DevicePolicyBridge(applicationContext)
MethodChannel(flutterEngine.dartExecutor.binaryMessenger, CHANNEL)
.setMethodCallHandler { call, result ->
when (call.method) {
"status" -> result.success(bridge.status())
"releaseDeviceOwner" -> result.success(bridge.releaseDeviceOwner())
else -> result.notImplemented()
}
}
}
companion object {
private const val CHANNEL = "com.kuhy.focus_owner/device_policy"
}
}

12
focus_owner/android/app/src/main/res/drawable-v21/launch_background.xml Normal file
View File

@ -0,0 +1,12 @@
<?xml version="1.0" encoding="utf-8"?>
<!-- Modify this file to customize your launch splash screen -->
<layer-list xmlns:android="http://schemas.android.com/apk/res/android">
<item android:drawable="?android:colorBackground" />
<!-- You can insert your own image assets here -->
<!-- <item>
<bitmap
android:gravity="center"
android:src="@mipmap/launch_image" />
</item> -->
</layer-list>

12
focus_owner/android/app/src/main/res/drawable/launch_background.xml Normal file
View File

@ -0,0 +1,12 @@
<?xml version="1.0" encoding="utf-8"?>
<!-- Modify this file to customize your launch splash screen -->
<layer-list xmlns:android="http://schemas.android.com/apk/res/android">
<item android:drawable="@android:color/white" />
<!-- You can insert your own image assets here -->
<!-- <item>
<bitmap
android:gravity="center"
android:src="@mipmap/launch_image" />
</item> -->
</layer-list>

18
focus_owner/android/app/src/main/res/values-night/styles.xml Normal file
View File

@ -0,0 +1,18 @@
<?xml version="1.0" encoding="utf-8"?>
<resources>
<!-- Theme applied to the Android Window while the process is starting when the OS's Dark Mode setting is on -->
<style name="LaunchTheme" parent="@android:style/Theme.Black.NoTitleBar">
<!-- Show a splash screen on the activity. Automatically removed when
the Flutter engine draws its first frame -->
<item name="android:windowBackground">@drawable/launch_background</item>
</style>
<!-- Theme applied to the Android Window as soon as the process has started.
This theme determines the color of the Android Window while your
Flutter UI initializes, as well as behind your Flutter UI while its
running.
This Theme is only used starting with V2 of Flutter's Android embedding. -->
<style name="NormalTheme" parent="@android:style/Theme.Black.NoTitleBar">
<item name="android:windowBackground">?android:colorBackground</item>
</style>
</resources>

18
focus_owner/android/app/src/main/res/values/styles.xml Normal file
View File

@ -0,0 +1,18 @@
<?xml version="1.0" encoding="utf-8"?>
<resources>
<!-- Theme applied to the Android Window while the process is starting when the OS's Dark Mode setting is off -->
<style name="LaunchTheme" parent="@android:style/Theme.Light.NoTitleBar">
<!-- Show a splash screen on the activity. Automatically removed when
the Flutter engine draws its first frame -->
<item name="android:windowBackground">@drawable/launch_background</item>
</style>
<!-- Theme applied to the Android Window as soon as the process has started.
This theme determines the color of the Android Window while your
Flutter UI initializes, as well as behind your Flutter UI while its
running.
This Theme is only used starting with V2 of Flutter's Android embedding. -->
<style name="NormalTheme" parent="@android:style/Theme.Light.NoTitleBar">
<item name="android:windowBackground">?android:colorBackground</item>
</style>
</resources>

12
focus_owner/android/app/src/main/res/xml/device_admin.xml Normal file
View File

@ -0,0 +1,12 @@
<?xml version="1.0" encoding="utf-8"?>
<!--
Device admin policy declaration.
Intentionally minimal: no <force-lock>, <wipe-data>, <disable-camera> or
password policies are requested. This build only needs the component to exist
so device owner can be provisioned and, more importantly, released again.
Capabilities are added alongside the enforcement that needs them.
-->
<device-admin xmlns:android="http://schemas.android.com/apk/res/android">
<uses-policies />
</device-admin>

7
focus_owner/android/app/src/profile/AndroidManifest.xml Normal file
View File

@ -0,0 +1,7 @@
<manifest xmlns:android="http://schemas.android.com/apk/res/android">
<!-- The INTERNET permission is required for development. Specifically,
the Flutter tool needs it to communicate with the running application
to allow setting breakpoints, to provide hot reload, etc.
-->
<uses-permission android:name="android.permission.INTERNET"/>
</manifest>

24
focus_owner/android/build.gradle.kts Normal file
View File

@ -0,0 +1,24 @@
allprojects {
repositories {
google()
mavenCentral()
}
}
val newBuildDir: Directory =
rootProject.layout.buildDirectory
.dir("../../build")
.get()
rootProject.layout.buildDirectory.value(newBuildDir)
subprojects {
val newSubprojectBuildDir: Directory = newBuildDir.dir(project.name)
project.layout.buildDirectory.value(newSubprojectBuildDir)
}
subprojects {
project.evaluationDependsOn(":app")
}
tasks.register<Delete>("clean") {
delete(rootProject.layout.buildDirectory)
}

6
focus_owner/android/gradle.properties Normal file
View File

@ -0,0 +1,6 @@
org.gradle.jvmargs=-Xmx8G -XX:MaxMetaspaceSize=4G -XX:ReservedCodeCacheSize=512m -XX:+HeapDumpOnOutOfMemoryError
android.useAndroidX=true
# This newDsl flag was added by the Flutter template
android.newDsl=false
# This builtInKotlin flag was added by the Flutter template
android.builtInKotlin=false

5
focus_owner/android/gradle/wrapper/gradle-wrapper.properties vendored Normal file
View File

@ -0,0 +1,5 @@
distributionBase=GRADLE_USER_HOME
distributionPath=wrapper/dists
zipStoreBase=GRADLE_USER_HOME
zipStorePath=wrapper/dists
distributionUrl=https\://services.gradle.org/distributions/gradle-9.1.0-all.zip

26
focus_owner/android/settings.gradle.kts Normal file
View File

@ -0,0 +1,26 @@
pluginManagement {
val flutterSdkPath =
run {
val properties = java.util.Properties()
file("local.properties").inputStream().use { properties.load(it) }
val flutterSdkPath = properties.getProperty("flutter.sdk")
require(flutterSdkPath != null) { "flutter.sdk not set in local.properties" }
flutterSdkPath
}
includeBuild("$flutterSdkPath/packages/flutter_tools/gradle")
repositories {
google()
mavenCentral()
gradlePluginPortal()
}
}
plugins {
id("dev.flutter.flutter-plugin-loader") version "1.0.0"
id("com.android.application") version "9.0.1" apply false
id("org.jetbrains.kotlin.android") version "2.3.20" apply false
}
include(":app")

61
focus_owner/lib/device_policy.dart Normal file
View File

@ -0,0 +1,61 @@
import 'package:flutter/services.dart';
/// Provisioning state reported by the native side.
class DevicePolicyStatus {
const DevicePolicyStatus({
required this.packageName,
required this.isDeviceOwner,
required this.isAdminActive,
required this.sdkInt,
required this.restrictionsApplied,
});
/// Builds a status from the raw platform-channel map.
factory DevicePolicyStatus.fromMap(Map<Object?, Object?> map) {
return DevicePolicyStatus(
packageName: map['packageName']! as String,
isDeviceOwner: map['isDeviceOwner']! as bool,
isAdminActive: map['isAdminActive']! as bool,
sdkInt: map['sdkInt']! as int,
restrictionsApplied: map['restrictionsApplied']! as bool,
);
}
final String packageName;
final bool isDeviceOwner;
final bool isAdminActive;
final int sdkInt;
/// Whether any user restriction is currently enforced.
///
/// Always false in this build: the app is provisioning-capable but inert
/// until the release path has been verified on a real device.
final bool restrictionsApplied;
}
/// Dart side of the device-policy platform channel.
class DevicePolicy {
const DevicePolicy([this.channel = _defaultChannel]);
static const MethodChannel _defaultChannel =
MethodChannel('com.kuhy.focus_owner/device_policy');
final MethodChannel channel;
/// Reads the current provisioning state.
Future<DevicePolicyStatus> status() async {
final result =
await channel.invokeMethod<Map<Object?, Object?>>('status');
return DevicePolicyStatus.fromMap(result!);
}
/// Relinquishes device ownership without wiping the device.
///
/// This is the escape hatch. It is deliberately reachable from the app's
/// main screen with no PC attached, because the situation it exists for is
/// one where ADB may already be disallowed.
Future<bool> releaseDeviceOwner() async {
final released = await channel.invokeMethod<bool>('releaseDeviceOwner');
return released ?? false;
}
}

223
focus_owner/lib/main.dart Normal file
View File

@ -0,0 +1,223 @@
import 'dart:async';
import 'package:flutter/material.dart';
import 'package:flutter/services.dart';
import 'package:focus_owner/device_policy.dart';
void main() => runApp(const FocusOwnerApp());
/// Shared palette: charcoal field with a single accent, matching the other
/// com.kuhy.* apps.
const Color _kField = Color(0xFF1B1D21);
const Color _kSurface = Color(0xFF24272C);
const Color _kAccent = Color(0xFF5B9DD9);
const Color _kText = Color(0xFFE8EAED);
const Color _kMuted = Color(0xFF9AA0A6);
const Color _kDanger = Color(0xFFD9776B);
const double _kGap = 16;
class FocusOwnerApp extends StatelessWidget {
const FocusOwnerApp({super.key});
@override
Widget build(BuildContext context) {
return MaterialApp(
title: 'Focus Owner',
debugShowCheckedModeBanner: false,
theme: ThemeData(
useMaterial3: true,
scaffoldBackgroundColor: _kField,
colorScheme: const ColorScheme.dark(
primary: _kAccent,
surface: _kSurface,
error: _kDanger,
),
),
home: const StatusPage(),
);
}
}
/// Shows provisioning state and offers the one action this build supports:
/// giving up device ownership.
class StatusPage extends StatefulWidget {
const StatusPage({super.key, this.policy = const DevicePolicy()});
final DevicePolicy policy;
@override
State<StatusPage> createState() => _StatusPageState();
}
class _StatusPageState extends State<StatusPage> {
DevicePolicyStatus? _status;
String? _error;
bool _busy = false;
@override
void initState() {
super.initState();
unawaited(_refresh());
}
Future<void> _refresh() async {
setState(() => _busy = true);
try {
final status = await widget.policy.status();
if (!mounted) return;
setState(() {
_status = status;
_error = null;
});
} on PlatformException catch (e) {
if (!mounted) return;
setState(() => _error = e.message ?? 'Could not read device policy state');
} finally {
if (mounted) setState(() => _busy = false);
}
}
Future<void> _release() async {
final confirmed = await showDialog<bool>(
context: context,
builder: (context) => AlertDialog(
backgroundColor: _kSurface,
title: const Text('Release device owner?'),
content: const Text(
'The app will give up device ownership. Enforcement stops and '
'ownership can only be restored by a factory reset.',
),
actions: [
TextButton(
onPressed: () => Navigator.pop(context, false),
child: const Text('Cancel'),
),
TextButton(
onPressed: () => Navigator.pop(context, true),
style: TextButton.styleFrom(foregroundColor: _kDanger),
child: const Text('Release'),
),
],
),
);
if (confirmed != true) return;
setState(() => _busy = true);
final released = await widget.policy.releaseDeviceOwner();
if (!mounted) return;
setState(() => _busy = false);
ScaffoldMessenger.of(context).showSnackBar(
SnackBar(
content: Text(
released ? 'Device owner released' : 'Release failed - see logcat',
),
),
);
await _refresh();
}
@override
Widget build(BuildContext context) {
final status = _status;
final error = _error;
final Widget body;
if (error != null) {
body = const _Message(text: 'Could not read state', color: _kDanger);
} else if (status != null) {
body = _StatusBody(status: status, busy: _busy, onRelease: _release);
} else {
body = const Center(child: CircularProgressIndicator());
}
return Scaffold(
appBar: AppBar(
backgroundColor: _kField,
foregroundColor: _kText,
title: const Text('Focus Owner'),
actions: [
IconButton(
onPressed: _busy ? null : _refresh,
icon: const Icon(Icons.refresh),
tooltip: 'Refresh',
),
],
),
body: Padding(padding: const EdgeInsets.all(_kGap), child: body),
);
}
}
class _StatusBody extends StatelessWidget {
const _StatusBody({
required this.status,
required this.busy,
required this.onRelease,
});
final DevicePolicyStatus status;
final bool busy;
final Future<void> Function() onRelease;
@override
Widget build(BuildContext context) {
return Column(
crossAxisAlignment: CrossAxisAlignment.start,
children: [
_Row(label: 'Package', value: status.packageName),
_Row(label: 'Device owner', value: status.isDeviceOwner ? 'yes' : 'no'),
_Row(label: 'Admin active', value: status.isAdminActive ? 'yes' : 'no'),
_Row(label: 'Android SDK', value: '${status.sdkInt}'),
_Row(
label: 'Restrictions',
value: status.restrictionsApplied ? 'applied' : 'none (inert build)',
),
const SizedBox(height: _kGap * 2),
if (status.isDeviceOwner)
FilledButton(
onPressed: busy ? null : onRelease,
style: FilledButton.styleFrom(backgroundColor: _kDanger),
child: const Text('Release device owner'),
)
else
const _Message(
text: 'Not provisioned as device owner. Nothing is enforced.',
color: _kMuted,
),
],
);
}
}
class _Row extends StatelessWidget {
const _Row({required this.label, required this.value});
final String label;
final String value;
@override
Widget build(BuildContext context) {
return Padding(
padding: const EdgeInsets.symmetric(vertical: _kGap / 2),
child: Row(
mainAxisAlignment: MainAxisAlignment.spaceBetween,
children: [
Text(label, style: const TextStyle(color: _kMuted)),
Text(value, style: const TextStyle(color: _kText)),
],
),
);
}
}
class _Message extends StatelessWidget {
const _Message({required this.text, required this.color});
final String text;
final Color color;
@override
Widget build(BuildContext context) =>
Text(text, style: TextStyle(color: color));
}

213
focus_owner/pubspec.lock Normal file
View File

@ -0,0 +1,213 @@
# Generated by pub
# See https://dart.dev/tools/pub/glossary#lockfile
packages:
async:
dependency: transitive
description:
name: async
sha256: e2eb0491ba5ddb6177742d2da23904574082139b07c1e33b8503b9f46f3e1a37
url: "https://pub.dev"
source: hosted
version: "2.13.1"
boolean_selector:
dependency: transitive
description:
name: boolean_selector
sha256: "8aab1771e1243a5063b8b0ff68042d67334e3feab9e95b9490f9a6ebf73b42ea"
url: "https://pub.dev"
source: hosted
version: "2.1.2"
characters:
dependency: transitive
description:
name: characters
sha256: faf38497bda5ead2a8c7615f4f7939df04333478bf32e4173fcb06d428b5716b
url: "https://pub.dev"
source: hosted
version: "1.4.1"
clock:
dependency: transitive
description:
name: clock
sha256: fddb70d9b5277016c77a80201021d40a2247104d9f4aa7bab7157b7e3f05b84b
url: "https://pub.dev"
source: hosted
version: "1.1.2"
collection:
dependency: transitive
description:
name: collection
sha256: "2f5709ae4d3d59dd8f7cd309b4e023046b57d8a6c82130785d2b0e5868084e76"
url: "https://pub.dev"
source: hosted
version: "1.19.1"
cupertino_icons:
dependency: "direct main"
description:
name: cupertino_icons
sha256: "41e005c33bd814be4d3096aff55b1908d419fde52ca656c8c47719ec745873cd"
url: "https://pub.dev"
source: hosted
version: "1.0.9"
fake_async:
dependency: transitive
description:
name: fake_async
sha256: "5368f224a74523e8d2e7399ea1638b37aecfca824a3cc4dfdf77bf1fa905ac44"
url: "https://pub.dev"
source: hosted
version: "1.3.3"
flutter:
dependency: "direct main"
description: flutter
source: sdk
version: "0.0.0"
flutter_lints:
dependency: "direct dev"
description:
name: flutter_lints
sha256: "3105dc8492f6183fb076ccf1f351ac3d60564bff92e20bfc4af9cc1651f4e7e1"
url: "https://pub.dev"
source: hosted
version: "6.0.0"
flutter_test:
dependency: "direct dev"
description: flutter
source: sdk
version: "0.0.0"
leak_tracker:
dependency: transitive
description:
name: leak_tracker
sha256: "33e2e26bdd85a0112ec15400c8cbffea70d0f9c3407491f672a2fad47915e2de"
url: "https://pub.dev"
source: hosted
version: "11.0.2"
leak_tracker_flutter_testing:
dependency: transitive
description:
name: leak_tracker_flutter_testing
sha256: "1dbc140bb5a23c75ea9c4811222756104fbcd1a27173f0c34ca01e16bea473c1"
url: "https://pub.dev"
source: hosted
version: "3.0.10"
leak_tracker_testing:
dependency: transitive
description:
name: leak_tracker_testing
sha256: "8d5a2d49f4a66b49744b23b018848400d23e54caf9463f4eb20df3eb8acb2eb1"
url: "https://pub.dev"
source: hosted
version: "3.0.2"
lints:
dependency: transitive
description:
name: lints
sha256: "12f842a479589fea194fe5c5a3095abc7be0c1f2ddfa9a0e76aed1dbd26a87df"
url: "https://pub.dev"
source: hosted
version: "6.1.0"
matcher:
dependency: transitive
description:
name: matcher
sha256: dc0b7dc7651697ea4ff3e69ef44b0407ea32c487a39fff6a4004fa585e901861
url: "https://pub.dev"
source: hosted
version: "0.12.19"
material_color_utilities:
dependency: transitive
description:
name: material_color_utilities
sha256: "9c337007e82b1889149c82ed242ed1cb24a66044e30979c44912381e9be4c48b"
url: "https://pub.dev"
source: hosted
version: "0.13.0"
meta:
dependency: transitive
description:
name: meta
sha256: "1741988757a65eb6b36abe716829688cf01910bbf91c34354ff7ec1c3de2b349"
url: "https://pub.dev"
source: hosted
version: "1.18.0"
path:
dependency: transitive
description:
name: path
sha256: "75cca69d1490965be98c73ceaea117e8a04dd21217b37b292c9ddbec0d955bc5"
url: "https://pub.dev"
source: hosted
version: "1.9.1"
sky_engine:
dependency: transitive
description: flutter
source: sdk
version: "0.0.0"
source_span:
dependency: transitive
description:
name: source_span
sha256: "56a02f1f4cd1a2d96303c0144c93bd6d909eea6bee6bf5a0e0b685edbd4c47ab"
url: "https://pub.dev"
source: hosted
version: "1.10.2"
stack_trace:
dependency: transitive
description:
name: stack_trace
sha256: "8b27215b45d22309b5cddda1aa2b19bdfec9df0e765f2de506401c071d38d1b1"
url: "https://pub.dev"
source: hosted
version: "1.12.1"
stream_channel:
dependency: transitive
description:
name: stream_channel
sha256: "969e04c80b8bcdf826f8f16579c7b14d780458bd97f56d107d3950fdbeef059d"
url: "https://pub.dev"
source: hosted
version: "2.1.4"
string_scanner:
dependency: transitive
description:
name: string_scanner
sha256: "921cd31725b72fe181906c6a94d987c78e3b98c2e205b397ea399d4054872b43"
url: "https://pub.dev"
source: hosted
version: "1.4.1"
term_glyph:
dependency: transitive
description:
name: term_glyph
sha256: "7f554798625ea768a7518313e58f83891c7f5024f88e46e7182a4558850a4b8e"
url: "https://pub.dev"
source: hosted
version: "1.2.2"
test_api:
dependency: transitive
description:
name: test_api
sha256: "949a932224383300f01be9221c39180316445ecb8e7547f70a41a35bf421fb9e"
url: "https://pub.dev"
source: hosted
version: "0.7.11"
vector_math:
dependency: transitive
description:
name: vector_math
sha256: d530bd74fea330e6e364cda7a85019c434070188383e1cd8d9777ee586914c5b
url: "https://pub.dev"
source: hosted
version: "2.2.0"
vm_service:
dependency: transitive
description:
name: vm_service
sha256: "0016aef94fc66495ac78af5859181e3f3bf2026bd8eecc72b9565601e19ab360"
url: "https://pub.dev"
source: hosted
version: "15.2.0"
sdks:
dart: ">=3.12.2 <4.0.0"
flutter: ">=3.18.0-18.0.pre.54"

89
focus_owner/pubspec.yaml Normal file
View File

@ -0,0 +1,89 @@
name: focus_owner
description: "A new Flutter project."
# The following line prevents the package from being accidentally published to
# pub.dev using `flutter pub publish`. This is preferred for private packages.
publish_to: 'none' # Remove this line if you wish to publish to pub.dev
# The following defines the version and build number for your application.
# A version number is three numbers separated by dots, like 1.2.43
# followed by an optional build number separated by a +.
# Both the version and the builder number may be overridden in flutter
# build by specifying --build-name and --build-number, respectively.
# In Android, build-name is used as versionName while build-number used as versionCode.
# Read more about Android versioning at https://developer.android.com/studio/publish/versioning
# In iOS, build-name is used as CFBundleShortVersionString while build-number is used as CFBundleVersion.
# Read more about iOS versioning at
# https://developer.apple.com/library/archive/documentation/General/Reference/InfoPlistKeyReference/Articles/CoreFoundationKeys.html
# In Windows, build-name is used as the major, minor, and patch parts
# of the product and file versions while build-number is used as the build suffix.
version: 1.0.0+1
environment:
sdk: ^3.12.2
# Dependencies specify other packages that your package needs in order to work.
# To automatically upgrade your package dependencies to the latest versions
# consider running `flutter pub upgrade --major-versions`. Alternatively,
# dependencies can be manually updated by changing the version numbers below to
# the latest version available on pub.dev. To see which dependencies have newer
# versions available, run `flutter pub outdated`.
dependencies:
flutter:
sdk: flutter
# The following adds the Cupertino Icons font to your application.
# Use with the CupertinoIcons class for iOS style icons.
cupertino_icons: ^1.0.8
dev_dependencies:
flutter_test:
sdk: flutter
# The "flutter_lints" package below contains a set of recommended lints to
# encourage good coding practices. The lint set provided by the package is
# activated in the `analysis_options.yaml` file located at the root of your
# package. See that file for information about deactivating specific lint
# rules and activating additional ones.
flutter_lints: ^6.0.0
# For information on the generic Dart part of this file, see the
# following page: https://dart.dev/tools/pub/pubspec
# The following section is specific to Flutter packages.
flutter:
# The following line ensures that the Material Icons font is
# included with your application, so that you can use the icons in
# the material Icons class.
uses-material-design: true
# To add assets to your application, add an assets section, like this:
# assets:
# - images/a_dot_burr.jpeg
# - images/a_dot_ham.jpeg
# An image asset can refer to one or more resolution-specific "variants", see
# https://flutter.dev/to/resolution-aware-images
# For details regarding adding assets from package dependencies, see
# https://flutter.dev/to/asset-from-package
# To add custom fonts to your application, add a fonts section here,
# in this "flutter" section. Each entry in this list should have a
# "family" key with the font family name, and a "fonts" key with a
# list giving the asset and other descriptors for the font. For
# example:
# fonts:
# - family: Schyler
# fonts:
# - asset: fonts/Schyler-Regular.ttf
# - asset: fonts/Schyler-Italic.ttf
# style: italic
# - family: Trajan Pro
# fonts:
# - asset: fonts/TrajanPro.ttf
# - asset: fonts/TrajanPro_Bold.ttf
# weight: 700
#
# For details regarding fonts from package dependencies,
# see https://flutter.dev/to/font-from-package

147
focus_owner/test/widget_test.dart Normal file
View File

@ -0,0 +1,147 @@
import 'package:flutter/material.dart';
import 'package:flutter/services.dart';
import 'package:flutter_test/flutter_test.dart';
import 'package:focus_owner/device_policy.dart';
import 'package:focus_owner/main.dart';
/// Builds a channel whose handler answers from [responses], recording calls.
({MethodChannel channel, List<MethodCall> calls}) _fakeChannel(
Map<String, Object?> responses,
) {
const channel = MethodChannel('test/device_policy');
final calls = <MethodCall>[];
TestDefaultBinaryMessengerBinding.instance.defaultBinaryMessenger
.setMockMethodCallHandler(channel, (call) async {
calls.add(call);
if (!responses.containsKey(call.method)) {
throw PlatformException(code: 'unimplemented', message: 'no stub');
}
return responses[call.method];
});
return (channel: channel, calls: calls);
}
Map<String, Object?> _statusMap({bool isDeviceOwner = false}) => {
'packageName': 'com.kuhy.focus_owner',
'isDeviceOwner': isDeviceOwner,
'isAdminActive': isDeviceOwner,
'sdkInt': 36,
'restrictionsApplied': false,
};
void main() {
group('DevicePolicyStatus', () {
test('parses the native status map', () {
final status = DevicePolicyStatus.fromMap(_statusMap(isDeviceOwner: true));
expect(status.packageName, 'com.kuhy.focus_owner');
expect(status.isDeviceOwner, isTrue);
expect(status.sdkInt, 36);
});
test('this build never reports restrictions as applied', () {
// The app is provisioning-capable but inert; enforcement ships only
// after the release path is verified on a device.
expect(
DevicePolicyStatus.fromMap(_statusMap()).restrictionsApplied,
isFalse,
);
});
});
group('DevicePolicy', () {
test('status() invokes the native status method', () async {
final fake = _fakeChannel({'status': _statusMap()});
final status = await DevicePolicy(fake.channel).status();
expect(fake.calls.single.method, 'status');
expect(status.isDeviceOwner, isFalse);
});
test('releaseDeviceOwner() forwards the native result', () async {
final fake = _fakeChannel({'releaseDeviceOwner': true});
expect(await DevicePolicy(fake.channel).releaseDeviceOwner(), isTrue);
expect(fake.calls.single.method, 'releaseDeviceOwner');
});
test('releaseDeviceOwner() reports failure rather than throwing', () async {
// The button that calls this is the last resort before a factory reset,
// so a failure must surface in the UI, never as an unhandled error.
final fake = _fakeChannel({'releaseDeviceOwner': false});
expect(await DevicePolicy(fake.channel).releaseDeviceOwner(), isFalse);
});
test('releaseDeviceOwner() treats a null reply as failure', () async {
final fake = _fakeChannel({'releaseDeviceOwner': null});
expect(await DevicePolicy(fake.channel).releaseDeviceOwner(), isFalse);
});
});
group('StatusPage', () {
testWidgets('hides the release button when not device owner',
(tester) async {
final fake = _fakeChannel({'status': _statusMap()});
await tester.pumpWidget(
MaterialApp(home: StatusPage(policy: DevicePolicy(fake.channel))),
);
await tester.pumpAndSettle();
expect(find.text('Release device owner'), findsNothing);
expect(find.textContaining('Nothing is enforced'), findsOneWidget);
});
testWidgets('shows the release button when device owner', (tester) async {
final fake = _fakeChannel({'status': _statusMap(isDeviceOwner: true)});
await tester.pumpWidget(
MaterialApp(home: StatusPage(policy: DevicePolicy(fake.channel))),
);
await tester.pumpAndSettle();
expect(find.text('Release device owner'), findsOneWidget);
expect(find.text('none (inert build)'), findsOneWidget);
});
testWidgets('release is confirmed before it runs', (tester) async {
final fake = _fakeChannel({
'status': _statusMap(isDeviceOwner: true),
'releaseDeviceOwner': true,
});
await tester.pumpWidget(
MaterialApp(home: StatusPage(policy: DevicePolicy(fake.channel))),
);
await tester.pumpAndSettle();
await tester.tap(find.text('Release device owner'));
await tester.pumpAndSettle();
expect(find.text('Release device owner?'), findsOneWidget);
// Cancelling must not call through to the platform.
await tester.tap(find.text('Cancel'));
await tester.pumpAndSettle();
expect(
fake.calls.where((c) => c.method == 'releaseDeviceOwner'),
isEmpty,
);
});
testWidgets('confirming release calls the platform', (tester) async {
final fake = _fakeChannel({
'status': _statusMap(isDeviceOwner: true),
'releaseDeviceOwner': true,
});
await tester.pumpWidget(
MaterialApp(home: StatusPage(policy: DevicePolicy(fake.channel))),
);
await tester.pumpAndSettle();
await tester.tap(find.text('Release device owner'));
await tester.pumpAndSettle();
await tester.tap(find.text('Release'));
await tester.pumpAndSettle();
expect(
fake.calls.where((c) => c.method == 'releaseDeviceOwner'),
hasLength(1),
);
expect(find.text('Device owner released'), findsOneWidget);
});
});
}

47
linux_configuration/tests/test_strip_workout_hosts.py
View File

@ -90,3 +90,50 @@ class TestMain:
result = dest.read_text(encoding="utf-8")
assert "www.youtube.com" not in result
assert "keepme.com" in result
class TestSubdomainMatching:
"""Regression tests for the exact-match bug fixed on 2026-08-05.
The original implementation tested ``name.lower() in unblock``, i.e. exact
set membership. YouTube streams video from dynamically named CDN hosts, so
an allowlist entry of ``googlevideo.com`` released nothing that actually
carried video and playback stayed blocked during workouts.
"""
def test_releases_dynamic_cdn_subdomains(self, tmp_path: Path) -> None:
"""A parent domain releases the CDN hosts beneath it."""
src = tmp_path / "hosts"
src.write_text(
"0.0.0.0 googlevideo.com\n"
"0.0.0.0 r1---sn-4g5e6nls.googlevideo.com\n"
"0.0.0.0 r1---sn-4g5lne7s.googlevideo.com\n"
"0.0.0.0 keepme.com\n",
encoding="utf-8",
)
dest = tmp_path / "out"
_strip(src, dest, frozenset({"googlevideo.com"}))
result = dest.read_text(encoding="utf-8")
assert "googlevideo.com" not in result
assert "keepme.com" in result
def test_does_not_release_lookalike_domains(self, tmp_path: Path) -> None:
"""Suffix matching must not be fooled by substring similarity."""
src = tmp_path / "hosts"
src.write_text(
"0.0.0.0 notgooglevideo.com\n0.0.0.0 googlevideo.com.evil.net\n",
encoding="utf-8",
)
dest = tmp_path / "out"
_strip(src, dest, frozenset({"googlevideo.com"}))
result = dest.read_text(encoding="utf-8")
assert "notgooglevideo.com" in result
assert "googlevideo.com.evil.net" in result
def test_case_and_trailing_dot_are_normalised(self, tmp_path: Path) -> None:
"""Host names are compared case-insensitively and without the root dot."""
src = tmp_path / "hosts"
src.write_text("0.0.0.0 WWW.YouTube.COM.\n", encoding="utf-8")
dest = tmp_path / "out"
_strip(src, dest, frozenset({"youtube.com"}))
assert dest.read_text(encoding="utf-8") == ""

17
phone_focus_mode/strip_workout_hosts.py
View File

@ -20,6 +20,21 @@ import sys
_MIN_HOSTS_FIELDS = 2
def _matches_unblock(name: str, unblock: frozenset[str]) -> bool:
"""Return whether ``name`` is an unblocked domain or a subdomain of one.
Suffix matching, not equality. YouTube serves video from dynamically named
CDN hosts such as ``r1---sn-4g5e6nls.googlevideo.com``; testing those for
exact membership against ``googlevideo.com`` never matches, so playback
stayed blocked during workouts even though the domain was allowlisted.
"""
candidate = name.lower().rstrip(".")
return any(
candidate == domain or candidate.endswith(f".{domain}")
for domain in (entry.lower().rstrip(".") for entry in unblock)
)
def _strip(source: Path, dest: Path, unblock: frozenset[str]) -> None:
"""Write ``source`` to ``dest`` minus lines that map an unblocked domain."""
text = source.read_text(encoding="utf-8", errors="replace")
@ -32,7 +47,7 @@ def _strip(source: Path, dest: Path, unblock: frozenset[str]) -> None:
parts = stripped.split()
names = parts[1:]
if len(parts) >= _MIN_HOSTS_FIELDS and any(
name.lower() in unblock for name in names
_matches_unblock(name, unblock) for name in names
):
continue
kept.append(line)

26
python_pkg/focus_policy/__init__.py Normal file
View File

@ -0,0 +1,26 @@
"""Device-independent focus-mode policy.
This package holds the *policy* half of phone focus mode: which packages are
allowed, how close to home counts as "at home", when the night curfew runs, and
which domains a workout temporarily unblocks.
It deliberately contains no enforcement code and no Android calls. The rooted
shell implementation (``phone_focus_mode/``) and the future unrooted Device
Owner app consume the same policy from here, so the two can never drift.
"""
from __future__ import annotations
from python_pkg.focus_policy.model import (
CurfewWindow,
FocusPolicy,
HomeLocation,
PolicyError,
)
__all__ = [
"CurfewWindow",
"FocusPolicy",
"HomeLocation",
"PolicyError",
]

170
python_pkg/focus_policy/blocklist.py Normal file
View File

@ -0,0 +1,170 @@
"""Turn a hosts-format blocklist into a filter-agnostic domain list.
The rooted enforcer ships a hosts file to the phone and bind-mounts it over
``/system/etc/hosts``. An unrooted phone cannot do that, so the same blocklist
has to be delivered as *domains* — to a DoT resolver, or to a local VpnService
that answers DNS itself. This module performs that conversion.
Two properties of the generated hosts file make a naive parser wrong, and both
are handled here:
1. ``generate_hosts_file.sh`` **comments out** deliberate unblocks rather than
deleting them (``#0.0.0.0 facebook.com``). Stripping ``#`` and taking field
two would silently re-block every site the user chose to allow.
2. The upstream StevenBlack header carries loopback and IPv6 housekeeping lines
(``127.0.0.1 localhost``, ``::1``, ``ff02::1``). Treating those as blocked
domains would blackhole ``localhost``.
"""
from __future__ import annotations
from dataclasses import dataclass
import re
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from collections.abc import Iterable, Iterator
# Only entries pointing at the null route are blocks. Anything else in a hosts
# file is either loopback housekeeping or a real mapping, never a block.
_IPV4_SINK = ".".join(["0"] * 4)
_IPV6_SINK = "::"
_BLOCK_SINKS = frozenset({_IPV4_SINK, _IPV6_SINK})
_MIN_HOSTS_FIELDS = 2
_MAX_DOMAIN_LEN = 253
# A hostname label: 1-63 chars, alphanumeric or hyphen, no leading/trailing
# hyphen. A domain is two or more labels joined by dots.
_VALID_DOMAIN = re.compile(
r"^(?!-)[a-z0-9-]{1,63}(?<!-)(\.(?!-)[a-z0-9-]{1,63}(?<!-))+$",
)
def is_valid_domain(candidate: str) -> bool:
"""Return whether ``candidate`` is a syntactically usable domain name.
The upstream StevenBlack list carries provenance annotations such as
``#[adware.zeno]`` and stray tokens like ``"spam``. Stripping the comment
marker turns those into plausible-looking names, and feeding them onward
produces a config the resolver rejects outright — dnsmasq fails to start on
``address=/"spam/``, which would take DNS down rather than block anything.
They also must not be honoured as unblocks, since ``#[cams.com]`` is an
annotation, not the user's decision to allow ``cams.com``.
"""
return len(candidate) <= _MAX_DOMAIN_LEN and bool(_VALID_DOMAIN.match(candidate))
@dataclass(frozen=True)
class BlocklistStats:
"""Counts describing one parse, for logging and sanity checks."""
total_lines: int
blocked_domains: int
commented_unblocks: int
skipped_non_sink: int
rejected_invalid: int
def _iter_entries(lines: Iterable[str]) -> Iterator[tuple[str, list[str], bool]]:
"""Yield ``(sink, names, was_commented)`` for every hosts-style entry.
Commented entries are yielded too, flagged, because a commented block is an
explicit *unblock* that must override a block of the same name.
"""
for line in lines:
stripped = line.strip()
if not stripped:
continue
commented = stripped.startswith("#")
body = stripped.lstrip("#").strip() if commented else stripped
if not body:
continue
parts = body.split()
if len(parts) < _MIN_HOSTS_FIELDS:
continue
yield parts[0], parts[1:], commented
def parse_hosts_blocklist(text: str) -> tuple[frozenset[str], BlocklistStats]:
"""Return the domains blocked by ``text``, plus statistics about the parse.
A domain that appears commented out anywhere is treated as unblocked, even
if an uncommented block for it also exists: the commented form is the
user's deliberate exception and must win.
"""
blocked: set[str] = set()
unblocked: set[str] = set()
total = skipped = rejected = 0
for sink, names, commented in _iter_entries(text.splitlines()):
total += 1
if sink not in _BLOCK_SINKS:
skipped += 1
continue
target = unblocked if commented else blocked
for name in names:
domain = name.lower().rstrip(".")
if not is_valid_domain(domain):
rejected += 1
continue
target.add(domain)
effective = frozenset(blocked - unblocked)
return effective, BlocklistStats(
total_lines=total,
blocked_domains=len(effective),
commented_unblocks=len(unblocked),
skipped_non_sink=skipped,
rejected_invalid=rejected,
)
def domain_matches(domain: str, patterns: Iterable[str]) -> bool:
"""Return whether ``domain`` equals or is a subdomain of any pattern.
Suffix matching is what makes a workout exception actually work. YouTube
serves video from dynamic CDN names like ``r1---sn-4g5e6nls.googlevideo.com``;
an exact-equality check against ``googlevideo.com`` never matches them, so
playback stays blocked. Comparing suffixes covers the whole domain tree.
"""
candidate = domain.lower().rstrip(".")
for pattern in patterns:
base = pattern.lower().rstrip(".")
if candidate == base or candidate.endswith(f".{base}"):
return True
return False
def apply_exceptions(
blocked: frozenset[str],
exceptions: Iterable[str],
) -> frozenset[str]:
"""Remove every domain covered by ``exceptions`` from ``blocked``.
Used for the workout unblock: while a workout is in progress the YouTube
domain tree is released. Unlike the shell implementation's exact-match
filter, this releases subdomains too.
"""
patterns = tuple(exceptions)
if not patterns:
return blocked
return frozenset(
domain for domain in blocked if not domain_matches(domain, patterns)
)
def to_dnsmasq_conf(domains: Iterable[str]) -> str:
"""Render domains as dnsmasq ``address=`` directives.
Suitable for the self-hosted DoT path, where dnsmasq is already the
resolver behind the TLS terminator. ``address=/domain/`` returns NXDOMAIN
for the domain and every subdomain, so the CDN-subdomain problem above
cannot recur at the resolver.
"""
return "".join(f"address=/{domain}/\n" for domain in sorted(domains))
def to_domain_list(domains: Iterable[str]) -> str:
"""Render domains one per line, for the VpnService fallback to load."""
return "".join(f"{domain}\n" for domain in sorted(domains))

60
python_pkg/focus_policy/export.py Normal file
View File

@ -0,0 +1,60 @@
"""Render a :class:`FocusPolicy` into a backend-neutral JSON document.
The Device Owner app is written in Kotlin and cannot read ``config.sh`` or
import this package. It reads this JSON instead, so both enforcement backends
stay driven by one policy definition.
Sorted keys and sorted lists keep the output stable, so committing a rendered
policy produces a meaningful diff when the policy actually changes rather than
noise from set iteration order.
"""
from __future__ import annotations
import json
from typing import TYPE_CHECKING, Any
if TYPE_CHECKING:
from python_pkg.focus_policy.model import FocusPolicy
SCHEMA_VERSION = 1
def policy_to_dict(policy: FocusPolicy) -> dict[str, Any]:
"""Return a JSON-serialisable representation of ``policy``."""
curfew: dict[str, str] | None = None
if policy.curfew is not None:
curfew = {
"start": policy.curfew.start.strftime("%H:%M"),
"end": policy.curfew.end.strftime("%H:%M"),
}
return {
"schema_version": SCHEMA_VERSION,
"home": {
"latitude": policy.home.latitude,
"longitude": policy.home.longitude,
"radius_m": policy.home.radius_m,
"hysteresis_m": policy.home.hysteresis_m,
},
"curfew": curfew,
"launcher_package": policy.launcher_package,
"allowed_packages": sorted(policy.allowed_packages),
"night_allowed_packages": sorted(policy.night_allowed_packages),
"never_disable_prefixes": sorted(policy.never_disable_prefixes),
"workout_unblock_domains": sorted(policy.workout_unblock_domains),
"browser_packages": sorted(policy.browser_packages),
}
def policy_to_json(policy: FocusPolicy, *, redact_home: bool = False) -> str:
"""Return ``policy`` as pretty-printed JSON.
``redact_home`` blanks the coordinates so a rendered policy can be attached
to a bug report or committed as a fixture without disclosing where the user
lives. The radius and hysteresis are kept, since they carry no location.
"""
payload = policy_to_dict(policy)
if redact_home:
payload["home"] = dict(payload["home"], latitude=None, longitude=None)
return json.dumps(payload, indent=2, sort_keys=True, ensure_ascii=False)

177
python_pkg/focus_policy/loader.py Normal file
View File

@ -0,0 +1,177 @@
"""Load the focus-mode policy from the shell config that already exists.
``phone_focus_mode/config.sh`` is the live source of truth for the rooted
enforcer. Rather than copy its lists into Python — where they would silently
drift the first time either side is edited — this module parses them.
Only simple ``export NAME="..."`` assignments are read. This is not a shell
interpreter: it deliberately understands the narrow subset ``config.sh`` uses
for policy, and raises rather than guessing at anything else.
"""
from __future__ import annotations
from datetime import time
import re
from typing import TYPE_CHECKING
from python_pkg.focus_policy.model import (
CurfewWindow,
FocusPolicy,
HomeLocation,
PolicyError,
)
if TYPE_CHECKING:
from pathlib import Path
# Matches `export NAME="value"` / `NAME='value'` / `NAME=bare`, including the
# multi-line quoted blocks config.sh uses for its package lists.
_ASSIGNMENT = re.compile(
r"^\s*(?:export\s+|readonly\s+)*(?P<name>[A-Za-z_][A-Za-z0-9_]*)="
r"(?P<value>\"[^\"]*\"|'[^']*'|\S*)",
re.MULTILINE | re.DOTALL,
)
_HHMM_LEN = 4
# A quoted value needs at least the opening and closing quote characters.
_MIN_QUOTED_LEN = 2
def parse_shell_assignments(text: str) -> dict[str, str]:
"""Return every simple assignment in ``text`` as a name -> value mapping.
Later assignments win, matching shell semantics where re-assigning a name
overwrites the earlier value.
"""
values: dict[str, str] = {}
for match in _ASSIGNMENT.finditer(text):
raw = match.group("value")
if (
raw[:1] in {'"', "'"}
and raw[-1:] == raw[:1]
and len(raw) >= _MIN_QUOTED_LEN
):
raw = raw[1:-1]
values[match.group("name")] = raw
return values
def parse_package_list(value: str) -> frozenset[str]:
"""Return the non-comment, non-blank entries of a whitespace-separated list."""
return frozenset(
line.strip()
for line in value.splitlines()
if line.strip() and not line.lstrip().startswith("#")
)
def parse_hhmm(value: str, *, field_name: str) -> time:
"""Parse a bare ``HHMM`` string as used by ``NIGHT_CURFEW_START``/``_END``."""
digits = value.strip()
if len(digits) != _HHMM_LEN or not digits.isdigit():
msg = f"{field_name} must be 4 digits (HHMM), got {value!r}"
raise PolicyError(msg)
hour, minute = int(digits[:2]), int(digits[2:])
try:
return time(hour=hour, minute=minute)
except ValueError as exc:
msg = f"{field_name} is not a valid time: {value!r}"
raise PolicyError(msg) from exc
def _require(values: dict[str, str], name: str) -> str:
"""Return ``values[name]`` or raise a PolicyError naming the missing key."""
try:
return values[name]
except KeyError as exc:
msg = f"required setting {name} is missing"
raise PolicyError(msg) from exc
def _coordinate(values: dict[str, str], name: str) -> float:
"""Return a coordinate, rejecting the shipped ``REDACTED_*`` placeholders.
``config_secrets.sh`` ships with placeholder coordinates so the real home
location never lands in git. Parsing one as a float raises, but the failure
reads as a generic type error; callers need to be told the file was never
filled in. Defaulting to 0.0 would be worse still — it silently relocates
"home" to the Atlantic and disables the location gate entirely.
"""
raw = _require(values, name).strip()
try:
return float(raw)
except ValueError as exc:
msg = (
f"{name} is {raw!r}, not a coordinate. Set the real value in "
"config_secrets.sh (untracked); the shipped file is a placeholder."
)
raise PolicyError(msg) from exc
def _float_setting(values: dict[str, str], name: str, default: float) -> float:
"""Return a numeric setting, falling back to ``default`` when unset."""
raw = values.get(name)
if raw is None or not raw.strip():
return default
try:
return float(raw)
except ValueError as exc:
msg = f"{name} must be numeric, got {raw!r}"
raise PolicyError(msg) from exc
def load_policy(
config_path: Path,
secrets_path: Path | None = None,
) -> FocusPolicy:
"""Build a :class:`FocusPolicy` from ``config.sh`` and its secrets file.
``HOME_LAT``/``HOME_LON`` live in ``config_secrets.sh``, which is untracked
because it pins down where the user lives. It is read from a separate path
and never embedded in this repository.
"""
values = parse_shell_assignments(config_path.read_text(encoding="utf-8"))
if secrets_path is None:
secrets_path = config_path.parent / "config_secrets.sh"
if secrets_path.is_file():
values.update(parse_shell_assignments(secrets_path.read_text(encoding="utf-8")))
home = HomeLocation(
latitude=_coordinate(values, "HOME_LAT"),
longitude=_coordinate(values, "HOME_LON"),
radius_m=_float_setting(values, "RADIUS", 150.0),
hysteresis_m=_float_setting(values, "HYSTERESIS", 30.0),
)
curfew: CurfewWindow | None = None
if values.get("NIGHT_CURFEW_ENABLED", "0").strip() == "1":
curfew = CurfewWindow(
start=parse_hhmm(
_require(values, "NIGHT_CURFEW_START"),
field_name="NIGHT_CURFEW_START",
),
end=parse_hhmm(
_require(values, "NIGHT_CURFEW_END"),
field_name="NIGHT_CURFEW_END",
),
)
launcher = values.get("LAUNCHER_PACKAGE", "").strip() or None
return FocusPolicy(
home=home,
allowed_packages=parse_package_list(_require(values, "WHITELIST")),
night_allowed_packages=parse_package_list(values.get("NIGHT_WHITELIST", "")),
never_disable_prefixes=tuple(
sorted(parse_package_list(values.get("SYSTEM_NEVER_DISABLE", "")))
),
workout_unblock_domains=parse_package_list(
values.get("WORKOUT_UNBLOCK_DOMAINS", "")
),
curfew=curfew,
launcher_package=launcher,
browser_packages=parse_package_list(values.get("BROWSER_PACKAGES", "")),
)

184
python_pkg/focus_policy/model.py Normal file
View File

@ -0,0 +1,184 @@
"""Typed policy model shared by every focus-mode enforcement backend.
The rooted implementation expresses all of this as exported shell strings in
``phone_focus_mode/config.sh``. That works for shell, but it cannot be unit
tested, and a Device Owner app written in Kotlin cannot read it at all. This
module is the language-neutral source of truth; ``export_json`` renders it into
a form any backend can load.
"""
from __future__ import annotations
from dataclasses import dataclass, field
import math
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from datetime import time
_MAX_LATITUDE = 90.0
_MAX_LONGITUDE = 180.0
_EARTH_RADIUS_M = 6_371_000.0
class PolicyError(ValueError):
"""Raised when a policy value is missing, malformed, or self-contradictory."""
@dataclass(frozen=True)
class HomeLocation:
"""The coordinate that focus mode is anchored to.
``radius_m`` is the distance at which restrictions switch on.
``hysteresis_m`` is added to the radius before restrictions switch back
*off*, so that GPS jitter at exactly the boundary cannot cause the enforcer
to flap between states many times a minute.
"""
latitude: float
longitude: float
radius_m: float = 150.0
hysteresis_m: float = 30.0
def __post_init__(self) -> None:
"""Reject coordinates and distances that cannot describe a real place."""
if not -_MAX_LATITUDE <= self.latitude <= _MAX_LATITUDE:
msg = f"latitude {self.latitude} outside [-90, 90]"
raise PolicyError(msg)
if not -_MAX_LONGITUDE <= self.longitude <= _MAX_LONGITUDE:
msg = f"longitude {self.longitude} outside [-180, 180]"
raise PolicyError(msg)
if self.radius_m <= 0:
msg = f"radius_m must be positive, got {self.radius_m}"
raise PolicyError(msg)
if self.hysteresis_m < 0:
msg = f"hysteresis_m must not be negative, got {self.hysteresis_m}"
raise PolicyError(msg)
def distance_m(self, latitude: float, longitude: float) -> float:
"""Return great-circle metres from home to the given point.
Mirrors the Haversine formula in ``focus_daemon.sh`` so that the Python
policy layer and the shell enforcer agree on the same boundary.
"""
lat1, lat2 = math.radians(self.latitude), math.radians(latitude)
delta_lat = math.radians(latitude - self.latitude)
delta_lon = math.radians(longitude - self.longitude)
haversine = (
math.sin(delta_lat / 2) ** 2
+ math.cos(lat1) * math.cos(lat2) * math.sin(delta_lon / 2) ** 2
)
return 2 * _EARTH_RADIUS_M * math.asin(math.sqrt(haversine))
def is_inside(
self,
latitude: float,
longitude: float,
*,
currently_focused: bool,
) -> bool:
"""Return whether this point counts as "at home", honouring hysteresis.
The threshold depends on the current state: leaving requires travelling
``hysteresis_m`` further than arriving did. Without this a reading that
hovers on the radius toggles enforcement on every poll.
"""
threshold = self.radius_m + (self.hysteresis_m if currently_focused else 0.0)
return self.distance_m(latitude, longitude) <= threshold
@dataclass(frozen=True)
class CurfewWindow:
"""A nightly window that wraps midnight (e.g. 23:00 -> 05:00)."""
start: time
end: time
def contains(self, moment: time) -> bool:
"""Return whether ``moment`` falls inside the window.
Handles both same-day windows (09:00-17:00) and the wrapping windows
focus mode actually uses (23:00-05:00), where "inside" means at or after
the start *or* strictly before the end.
"""
if self.start <= self.end:
return self.start <= moment < self.end
return moment >= self.start or moment < self.end
@dataclass(frozen=True)
class FocusPolicy:
"""The complete, backend-independent focus-mode policy."""
home: HomeLocation
allowed_packages: frozenset[str]
night_allowed_packages: frozenset[str]
never_disable_prefixes: tuple[str, ...]
workout_unblock_domains: frozenset[str] = frozenset()
curfew: CurfewWindow | None = None
launcher_package: str | None = None
browser_packages: frozenset[str] = field(default_factory=frozenset)
def __post_init__(self) -> None:
"""Reject policies that would lock the user out of the device."""
if self.launcher_package and self.launcher_package not in self.allowed_packages:
msg = (
f"launcher {self.launcher_package!r} is not in allowed_packages; "
"enforcing this policy would leave the device with no home screen"
)
raise PolicyError(msg)
orphans = self.night_allowed_packages - self.allowed_packages
if orphans:
msg = (
"night_allowed_packages must be a subset of allowed_packages; "
f"unknown at day level: {sorted(orphans)}"
)
raise PolicyError(msg)
def is_protected(self, package: str) -> bool:
"""Return whether a package must never be disabled.
Guards the system packages whose loss would brick core functions —
dialer, settings, IME, SystemUI. Prefix-matched, mirroring the shell
implementation, so that ``com.android.providers.telephony`` is covered
by the ``com.android.providers`` entry.
"""
return any(
package == prefix or package.startswith(f"{prefix}.")
for prefix in self.never_disable_prefixes
)
def is_allowed(self, package: str, *, during_curfew: bool = False) -> bool:
"""Return whether a package may run under the given conditions.
Protected system packages are always allowed. During curfew the much
smaller ``night_allowed_packages`` set applies instead of the day list.
"""
if self.is_protected(package):
return True
allowed = (
self.night_allowed_packages if during_curfew else self.allowed_packages
)
return package in allowed
def is_curfew_active(self, moment: time) -> bool:
"""Return whether the night curfew is in force at ``moment``."""
return self.curfew is not None and self.curfew.contains(moment)
def packages_to_block(
self,
installed: frozenset[str],
*,
during_curfew: bool = False,
) -> frozenset[str]:
"""Return the installed packages that this policy blocks.
Callers pass the set actually present on the device, so a package that
is not installed never appears in the result.
"""
return frozenset(
package
for package in installed
if not self.is_allowed(package, during_curfew=during_curfew)
)

1
python_pkg/focus_policy/tests/__init__.py Normal file
View File

@ -0,0 +1 @@
"""Tests for the focus-mode policy package."""

215
python_pkg/focus_policy/tests/test_blocklist.py Normal file
View File

@ -0,0 +1,215 @@
"""Tests for hosts-file parsing and blocklist rendering.
The cases here are drawn from real defects found in the live 194k-line
blocklist, not from imagined inputs: commented unblocks, upstream annotation
junk, and CDN subdomains that an exact-match exception filter misses.
"""
from __future__ import annotations
from typing import TYPE_CHECKING
import pytest
if TYPE_CHECKING:
from collections.abc import Callable, Iterable
from python_pkg.focus_policy.blocklist import (
apply_exceptions,
domain_matches,
is_valid_domain,
parse_hosts_blocklist,
to_dnsmasq_conf,
to_domain_list,
)
class TestIsValidDomain:
"""Domain syntax validation."""
@pytest.mark.parametrize(
"domain",
[
"example.com",
"a.b.c.example.com",
"xn--80ak6aa92e.com",
"r1---sn-4g5e6nls.googlevideo.com",
"1-2.co",
],
)
def test_accepts_real_domains(self, domain: str) -> None:
"""Ordinary names, punycode, and hyphen-heavy CDN names are valid."""
assert is_valid_domain(domain)
@pytest.mark.parametrize(
"token",
[
'"spam',
"#:",
"#[adware.zeno]",
"localhost",
"",
"-leading.com",
"trailing-.com",
"no_underscores.com",
"double..dot.com",
],
)
def test_rejects_junk_and_single_labels(self, token: str) -> None:
"""Annotation junk and bare single labels are not usable domains.
``address=/"spam/`` makes dnsmasq refuse to start, which would take DNS
down entirely rather than block anything.
"""
assert not is_valid_domain(token)
def test_rejects_overlong_name(self) -> None:
"""Names beyond the 253-character DNS limit are rejected."""
assert not is_valid_domain(("a" * 63 + ".") * 4 + "com")
class TestParseHostsBlocklist:
"""Extraction of blocked domains from hosts-format text."""
def test_extracts_null_routed_domains(self) -> None:
"""Entries pointing at 0.0.0.0 are blocks."""
blocked, stats = parse_hosts_blocklist(
"0.0.0.0 ads.example\n0.0.0.0 tracker.example\n",
)
assert blocked == frozenset({"ads.example", "tracker.example"})
assert stats.blocked_domains == 2
def test_commented_entry_unblocks(self) -> None:
"""A commented block is a deliberate exception and must win.
``generate_hosts_file.sh`` comments unblocks out rather than deleting
them, so a parser that merely strips ``#`` re-blocks every site the
user chose to allow.
"""
blocked, stats = parse_hosts_blocklist(
"0.0.0.0 facebook.com\n#0.0.0.0 facebook.com\n0.0.0.0 ads.example\n",
)
assert blocked == frozenset({"ads.example"})
assert stats.commented_unblocks == 1
def test_ignores_loopback_and_ipv6_housekeeping(self) -> None:
"""Non-sink mappings are not blocks; blackholing localhost breaks apps."""
blocked, stats = parse_hosts_blocklist(
"127.0.0.1 localhost\n::1 localhost\nff02::1 ip6-allnodes\n"
"255.255.255.255 broadcasthost\n0.0.0.0 ads.example\n",
)
assert blocked == frozenset({"ads.example"})
assert stats.skipped_non_sink == 4
def test_rejects_annotation_junk(self) -> None:
"""Upstream provenance annotations never become domains or unblocks."""
blocked, stats = parse_hosts_blocklist(
'0.0.0.0 ads.example\n#[adware.zeno]\n0.0.0.0 "spam\n',
)
assert blocked == frozenset({"ads.example"})
assert stats.rejected_invalid == 1
def test_annotation_does_not_unblock_a_real_domain(self) -> None:
"""``#[cams.com]`` is an annotation, not permission to allow cams.com."""
blocked, _ = parse_hosts_blocklist("0.0.0.0 cams.com\n#[cams.com]\n")
assert "cams.com" in blocked
def test_multiple_names_on_one_line(self) -> None:
"""A single sink line may list several hostnames."""
blocked, _ = parse_hosts_blocklist("0.0.0.0 a.example b.example\n")
assert blocked == frozenset({"a.example", "b.example"})
def test_normalises_case_and_trailing_dot(self) -> None:
"""Names are compared lowercased and without the root dot."""
blocked, _ = parse_hosts_blocklist("0.0.0.0 ADS.Example.\n")
assert blocked == frozenset({"ads.example"})
def test_ipv6_sink_counts_as_a_block(self) -> None:
"""``::`` is the IPv6 null route, equivalent to 0.0.0.0."""
blocked, _ = parse_hosts_blocklist(":: ads.example\n")
assert blocked == frozenset({"ads.example"})
@pytest.mark.parametrize("text", ["", "\n\n", "#\n", " \n", "0.0.0.0\n"])
def test_degenerate_input_yields_nothing(self, text: str) -> None:
"""Blank lines, bare comments, and sink-only lines produce no domains."""
blocked, _ = parse_hosts_blocklist(text)
assert blocked == frozenset()
class TestDomainMatches:
"""Suffix matching used by the workout exception."""
@pytest.mark.parametrize(
"domain",
[
"googlevideo.com",
"r1---sn-4g5e6nls.googlevideo.com",
"manifest.googlevideo.com",
],
)
def test_matches_domain_and_subdomains(self, domain: str) -> None:
"""A pattern covers its whole subtree.
YouTube serves video from dynamic ``r*---sn-*.googlevideo.com`` hosts;
exact equality against ``googlevideo.com`` never matches them, which is
why the shell implementation leaves playback blocked during workouts.
"""
assert domain_matches(domain, ["googlevideo.com"])
@pytest.mark.parametrize(
"domain",
["notgooglevideo.com", "googlevideo.com.evil.net", "example.com"],
)
def test_rejects_lookalikes(self, domain: str) -> None:
"""Suffix matching must not be fooled by substring similarity."""
assert not domain_matches(domain, ["googlevideo.com"])
def test_empty_pattern_list_matches_nothing(self) -> None:
"""No patterns means no matches."""
assert not domain_matches("example.com", [])
class TestApplyExceptions:
"""Releasing a domain subtree from the blocklist."""
def test_releases_subtree(self) -> None:
"""Exceptions remove the domain and everything under it."""
blocked = frozenset(
{"googlevideo.com", "r1---sn-x.googlevideo.com", "ads.example"},
)
assert apply_exceptions(blocked, ["googlevideo.com"]) == frozenset(
{"ads.example"},
)
def test_unrelated_blocks_survive(self) -> None:
"""An exception must not widen into unrelated domains."""
blocked = frozenset({"pornhub.com", "doubleclick.net", "youtube.com"})
assert apply_exceptions(blocked, ["youtube.com"]) == frozenset(
{"pornhub.com", "doubleclick.net"},
)
def test_no_exceptions_returns_input(self) -> None:
"""An empty exception list leaves the blocklist untouched."""
blocked = frozenset({"ads.example"})
assert apply_exceptions(blocked, []) == blocked
class TestRenderers:
"""Output formats consumed by the two delivery backends."""
def test_dnsmasq_conf_is_sorted_and_wildcarding(self) -> None:
"""``address=/domain/`` NXDOMAINs the domain and every subdomain."""
conf = to_dnsmasq_conf({"b.example", "a.example"})
assert conf == "address=/a.example/\naddress=/b.example/\n"
def test_domain_list_is_sorted_one_per_line(self) -> None:
"""The VpnService fallback loads a plain sorted list."""
assert to_domain_list({"b.example", "a.example"}) == "a.example\nb.example\n"
@pytest.mark.parametrize("renderer", [to_dnsmasq_conf, to_domain_list])
def test_empty_input_renders_empty(
self,
renderer: Callable[[Iterable[str]], str],
) -> None:
"""No domains renders an empty document, not a malformed one."""
assert renderer(frozenset()) == ""

243
python_pkg/focus_policy/tests/test_loader.py Normal file
View File

@ -0,0 +1,243 @@
"""Tests for parsing policy out of the existing shell config."""
from __future__ import annotations
from datetime import time
from typing import TYPE_CHECKING
import pytest
from python_pkg.focus_policy.export import policy_to_dict, policy_to_json
from python_pkg.focus_policy.loader import (
load_policy,
parse_hhmm,
parse_package_list,
parse_shell_assignments,
)
from python_pkg.focus_policy.model import PolicyError
if TYPE_CHECKING:
from pathlib import Path
MINIMAL_CONFIG = """
export RADIUS=150
export HYSTERESIS=30
export NIGHT_CURFEW_ENABLED=1
export NIGHT_CURFEW_START="2300"
export NIGHT_CURFEW_END="0500"
export LAUNCHER_PACKAGE="com.launcher"
export WHITELIST="
com.launcher
com.good
pl.mbank
"
export NIGHT_WHITELIST="
pl.mbank
"
export SYSTEM_NEVER_DISABLE="
com.android.settings
"
export WORKOUT_UNBLOCK_DOMAINS="
youtube.com
"
"""
SECRETS = "export HOME_LAT=52.2297\nexport HOME_LON=21.0122\n"
@pytest.fixture
def config_dir(tmp_path: Path) -> Path:
"""Write a minimal config plus secrets into a temporary directory."""
(tmp_path / "config.sh").write_text(MINIMAL_CONFIG, encoding="utf-8")
(tmp_path / "config_secrets.sh").write_text(SECRETS, encoding="utf-8")
return tmp_path
class TestParseShellAssignments:
"""The narrow shell-assignment subset the loader understands."""
@pytest.mark.parametrize(
("text", "expected"),
[
('export A="x"', {"A": "x"}),
("export A='x'", {"A": "x"}),
("A=x", {"A": "x"}),
("readonly A=x", {"A": "x"}),
('export A="one\ntwo"', {"A": "one\ntwo"}),
],
)
def test_assignment_forms(self, text: str, expected: dict[str, str]) -> None:
"""Quoted, bare, exported, readonly, and multi-line forms all parse."""
assert parse_shell_assignments(text) == expected
def test_later_assignment_wins(self) -> None:
"""Re-assignment overwrites, matching shell semantics."""
assert parse_shell_assignments("A=1\nA=2")["A"] == "2"
def test_ignores_prose(self) -> None:
"""Comment prose without an assignment yields nothing."""
assert parse_shell_assignments("# just a comment\n") == {}
class TestParsePackageList:
"""Whitespace-separated list blocks."""
def test_drops_blanks_and_comments(self) -> None:
"""Only real entries survive."""
assert parse_package_list("\ncom.a\n# note\n\n com.b \n") == frozenset(
{"com.a", "com.b"},
)
def test_empty_string_is_empty_set(self) -> None:
"""An unset list is empty, not a set containing an empty string."""
assert parse_package_list("") == frozenset()
class TestParseHhmm:
"""HHMM time parsing."""
def test_parses_valid_time(self) -> None:
"""A well-formed HHMM string becomes a time."""
assert parse_hhmm("2300", field_name="X") == time(23, 0)
@pytest.mark.parametrize("value", ["230", "23000", "abcd", "2560", "2400"])
def test_rejects_malformed_time(self, value: str) -> None:
"""Wrong length, non-digits, and impossible clock values all raise."""
with pytest.raises(PolicyError, match="X"):
parse_hhmm(value, field_name="X")
class TestLoadPolicy:
"""End-to-end policy loading."""
def test_loads_expected_policy(self, config_dir: Path) -> None:
"""Every field is populated from the two files."""
policy = load_policy(config_dir / "config.sh")
assert policy.home.latitude == pytest.approx(52.2297)
assert policy.home.radius_m == pytest.approx(150.0)
assert policy.allowed_packages == frozenset(
{"com.launcher", "com.good", "pl.mbank"},
)
assert policy.night_allowed_packages == frozenset({"pl.mbank"})
assert policy.never_disable_prefixes == ("com.android.settings",)
assert policy.workout_unblock_domains == frozenset({"youtube.com"})
assert policy.curfew is not None
assert policy.curfew.start == time(23, 0)
assert policy.launcher_package == "com.launcher"
def test_explicit_secrets_path_is_honoured(
self,
config_dir: Path,
tmp_path: Path,
) -> None:
"""Secrets may live outside the config directory."""
elsewhere = tmp_path / "other_secrets.sh"
elsewhere.write_text("export HOME_LAT=1.0\nexport HOME_LON=2.0\n", "utf-8")
policy = load_policy(config_dir / "config.sh", elsewhere)
assert policy.home.latitude == pytest.approx(1.0)
def test_rejects_placeholder_coordinates(self, config_dir: Path) -> None:
"""The shipped placeholder must fail loudly, never default to (0, 0).
Silently treating an unset coordinate as the origin would place "home"
in the Atlantic and disable the location gate entirely.
"""
(config_dir / "config_secrets.sh").write_text(
"export HOME_LAT=REDACTED_LAT\nexport HOME_LON=REDACTED_LON\n",
encoding="utf-8",
)
with pytest.raises(PolicyError, match="placeholder"):
load_policy(config_dir / "config.sh")
def test_missing_secrets_file_reports_missing_setting(
self,
config_dir: Path,
) -> None:
"""With no secrets at all the error names the absent key."""
(config_dir / "config_secrets.sh").unlink()
with pytest.raises(PolicyError, match="HOME_LAT"):
load_policy(config_dir / "config.sh")
def test_missing_whitelist_is_fatal(self, tmp_path: Path) -> None:
"""A config without an allowlist cannot produce a usable policy."""
(tmp_path / "config.sh").write_text("export RADIUS=150\n", encoding="utf-8")
(tmp_path / "config_secrets.sh").write_text(SECRETS, encoding="utf-8")
with pytest.raises(PolicyError, match="WHITELIST"):
load_policy(tmp_path / "config.sh")
def test_curfew_disabled_yields_no_window(self, config_dir: Path) -> None:
"""NIGHT_CURFEW_ENABLED=0 disables the window entirely."""
text = MINIMAL_CONFIG.replace(
"NIGHT_CURFEW_ENABLED=1",
"NIGHT_CURFEW_ENABLED=0",
)
(config_dir / "config.sh").write_text(text, encoding="utf-8")
assert load_policy(config_dir / "config.sh").curfew is None
def test_defaults_apply_when_numbers_absent(self, config_dir: Path) -> None:
"""Radius and hysteresis fall back to the documented defaults."""
text = MINIMAL_CONFIG.replace("export RADIUS=150\n", "").replace(
"export HYSTERESIS=30\n",
"",
)
(config_dir / "config.sh").write_text(text, encoding="utf-8")
policy = load_policy(config_dir / "config.sh")
assert policy.home.radius_m == pytest.approx(150.0)
assert policy.home.hysteresis_m == pytest.approx(30.0)
def test_blank_number_falls_back_to_default(self, config_dir: Path) -> None:
"""An empty value is treated as unset rather than as zero."""
text = MINIMAL_CONFIG.replace("export RADIUS=150", 'export RADIUS=""')
(config_dir / "config.sh").write_text(text, encoding="utf-8")
assert load_policy(config_dir / "config.sh").home.radius_m == pytest.approx(
150.0,
)
def test_non_numeric_radius_is_fatal(self, config_dir: Path) -> None:
"""A malformed number is an error, not a silent fallback."""
text = MINIMAL_CONFIG.replace("export RADIUS=150", "export RADIUS=wide")
(config_dir / "config.sh").write_text(text, encoding="utf-8")
with pytest.raises(PolicyError, match="RADIUS"):
load_policy(config_dir / "config.sh")
def test_blank_launcher_becomes_none(self, config_dir: Path) -> None:
"""An empty launcher setting disables the launcher invariant."""
text = MINIMAL_CONFIG.replace(
'export LAUNCHER_PACKAGE="com.launcher"',
'export LAUNCHER_PACKAGE=""',
)
(config_dir / "config.sh").write_text(text, encoding="utf-8")
assert load_policy(config_dir / "config.sh").launcher_package is None
class TestExport:
"""Rendering the policy for the Kotlin backend."""
def test_dict_round_trips_key_fields(self, config_dir: Path) -> None:
"""The exported document carries the policy the loader produced."""
payload = policy_to_dict(load_policy(config_dir / "config.sh"))
assert payload["schema_version"] == 1
assert payload["curfew"] == {"start": "23:00", "end": "05:00"}
assert payload["allowed_packages"] == ["com.good", "com.launcher", "pl.mbank"]
assert payload["home"]["latitude"] == pytest.approx(52.2297)
def test_absent_curfew_serialises_as_null(self, config_dir: Path) -> None:
"""No curfew renders as JSON null rather than being omitted."""
text = MINIMAL_CONFIG.replace(
"NIGHT_CURFEW_ENABLED=1",
"NIGHT_CURFEW_ENABLED=0",
)
(config_dir / "config.sh").write_text(text, encoding="utf-8")
assert policy_to_dict(load_policy(config_dir / "config.sh"))["curfew"] is None
def test_redaction_hides_coordinates_only(self, config_dir: Path) -> None:
"""Redacted output keeps the radius but drops where the user lives."""
policy = load_policy(config_dir / "config.sh")
assert "52.2297" not in policy_to_json(policy, redact_home=True)
assert "52.2297" in policy_to_json(policy)
assert '"radius_m": 150.0' in policy_to_json(policy, redact_home=True)
def test_json_is_stable(self, config_dir: Path) -> None:
"""Repeated renders are byte-identical, so diffs stay meaningful."""
policy = load_policy(config_dir / "config.sh")
assert policy_to_json(policy) == policy_to_json(policy)

226
python_pkg/focus_policy/tests/test_model.py Normal file
View File

@ -0,0 +1,226 @@
"""Tests for the policy model: geometry, curfew windows, and allow decisions."""
from __future__ import annotations
from datetime import time
import pytest
from python_pkg.focus_policy.model import (
CurfewWindow,
FocusPolicy,
HomeLocation,
PolicyError,
)
HOME = HomeLocation(latitude=52.2297, longitude=21.0122)
METRES_PER_DEGREE_LAT = 111_320.0
def _policy(
*,
allowed_packages: frozenset[str] | None = None,
night_allowed_packages: frozenset[str] | None = None,
curfew: CurfewWindow | None = None,
launcher_package: str | None = None,
) -> FocusPolicy:
"""Build a small policy, overriding individual fields per test."""
return FocusPolicy(
home=HOME,
allowed_packages=(
frozenset({"com.good", "pl.mbank", "com.launcher"})
if allowed_packages is None
else allowed_packages
),
night_allowed_packages=(
frozenset({"pl.mbank"})
if night_allowed_packages is None
else night_allowed_packages
),
never_disable_prefixes=("com.android.providers", "com.android.settings"),
curfew=curfew,
launcher_package=launcher_package,
)
class TestHomeLocation:
"""Coordinate validation and distance maths."""
@pytest.mark.parametrize(
("kwargs", "expected"),
[
({"latitude": 91.0, "longitude": 0.0}, "latitude"),
({"latitude": -91.0, "longitude": 0.0}, "latitude"),
({"latitude": 0.0, "longitude": 181.0}, "longitude"),
({"latitude": 0.0, "longitude": -181.0}, "longitude"),
({"latitude": 0.0, "longitude": 0.0, "radius_m": 0.0}, "radius_m"),
({"latitude": 0.0, "longitude": 0.0, "hysteresis_m": -1.0}, "hysteresis_m"),
],
)
def test_rejects_impossible_values(
self,
kwargs: dict[str, float],
expected: str,
) -> None:
"""Out-of-range coordinates and distances raise, naming the bad field."""
with pytest.raises(PolicyError, match=expected):
HomeLocation(**kwargs)
def test_accepts_boundary_coordinates(self) -> None:
"""The poles and the antimeridian are valid places."""
assert HomeLocation(latitude=90.0, longitude=180.0).latitude == 90.0
assert HomeLocation(latitude=-90.0, longitude=-180.0).longitude == -180.0
def test_distance_to_self_is_zero(self) -> None:
"""A point measured against itself is zero metres away."""
assert HOME.distance_m(HOME.latitude, HOME.longitude) == pytest.approx(0.0)
def test_distance_matches_known_offset(self) -> None:
"""One degree of latitude is ~111 km; used to pin the Haversine maths."""
away = HOME.distance_m(HOME.latitude + 1.0, HOME.longitude)
assert away == pytest.approx(METRES_PER_DEGREE_LAT, rel=0.01)
def test_inside_within_radius_regardless_of_state(self) -> None:
"""Well inside the radius is "home" whether or not focus is already on."""
near = HOME.latitude + 50 / METRES_PER_DEGREE_LAT
assert HOME.is_inside(near, HOME.longitude, currently_focused=False)
assert HOME.is_inside(near, HOME.longitude, currently_focused=True)
def test_hysteresis_prevents_flapping_at_the_boundary(self) -> None:
"""Between radius and radius+hysteresis the answer depends on state.
This is the whole point of hysteresis: arriving has not yet triggered
focus, but leaving has not yet released it, so GPS jitter around the
boundary cannot toggle enforcement on every poll.
"""
edge = HOME.latitude + 160 / METRES_PER_DEGREE_LAT
assert not HOME.is_inside(edge, HOME.longitude, currently_focused=False)
assert HOME.is_inside(edge, HOME.longitude, currently_focused=True)
def test_outside_hysteresis_band_is_away_in_both_states(self) -> None:
"""Beyond radius+hysteresis the user is away regardless of state."""
far = HOME.latitude + 500 / METRES_PER_DEGREE_LAT
assert not HOME.is_inside(far, HOME.longitude, currently_focused=False)
assert not HOME.is_inside(far, HOME.longitude, currently_focused=True)
class TestCurfewWindow:
"""Window containment, including the midnight wrap."""
@pytest.mark.parametrize(
("moment", "expected"),
[
(time(22, 59), False),
(time(23, 0), True),
(time(23, 59), True),
(time(0, 0), True),
(time(4, 59), True),
(time(5, 0), False),
(time(12, 0), False),
],
)
def test_wrapping_window_spans_midnight(
self,
moment: time,
*,
expected: bool,
) -> None:
"""23:00-05:00 covers the late evening and the small hours alike."""
window = CurfewWindow(start=time(23, 0), end=time(5, 0))
assert window.contains(moment) is expected
@pytest.mark.parametrize(
("moment", "expected"),
[
(time(8, 59), False),
(time(9, 0), True),
(time(16, 59), True),
(time(17, 0), False),
],
)
def test_same_day_window(self, moment: time, *, expected: bool) -> None:
"""A non-wrapping window is a plain half-open interval."""
window = CurfewWindow(start=time(9, 0), end=time(17, 0))
assert window.contains(moment) is expected
class TestFocusPolicy:
"""Allow/deny decisions and the invariants that keep the phone usable."""
def test_rejects_launcher_outside_allowlist(self) -> None:
"""A policy that blocks its own launcher would leave no home screen."""
with pytest.raises(PolicyError, match="no home screen"):
_policy(launcher_package="com.missing")
def test_accepts_launcher_inside_allowlist(self) -> None:
"""The guard permits a launcher that is actually allowed."""
assert _policy(launcher_package="com.launcher").launcher_package
def test_rejects_night_packages_absent_from_day_list(self) -> None:
"""A night allowance for a day-blocked package is contradictory."""
with pytest.raises(PolicyError, match="subset"):
_policy(night_allowed_packages=frozenset({"com.unknown"}))
@pytest.mark.parametrize(
("package", "expected"),
[
("com.android.settings", True),
("com.android.providers", True),
("com.android.providers.telephony", True),
("com.android.providersomething", False),
("com.good", False),
],
)
def test_protection_matches_whole_labels_only(
self,
package: str,
*,
expected: bool,
) -> None:
"""Prefix protection covers subpackages but not lookalike names.
``com.android.providers.telephony`` must be protected; the unrelated
``com.android.providersomething`` must not be swept in by a bare
string prefix test.
"""
assert _policy().is_protected(package) is expected
def test_protected_packages_are_allowed_even_during_curfew(self) -> None:
"""Core system packages survive the strictest state."""
policy = _policy()
assert policy.is_allowed("com.android.settings", during_curfew=True)
def test_curfew_narrows_the_allowlist(self) -> None:
"""Day-allowed apps can still be blocked at night."""
policy = _policy()
assert policy.is_allowed("com.good")
assert not policy.is_allowed("com.good", during_curfew=True)
assert policy.is_allowed("pl.mbank", during_curfew=True)
def test_unknown_package_is_blocked(self) -> None:
"""Anything not explicitly allowed is denied."""
assert not _policy().is_allowed("com.evil")
def test_is_curfew_active_without_window_is_false(self) -> None:
"""A policy with no curfew is never in curfew."""
assert not _policy().is_curfew_active(time(23, 30))
def test_is_curfew_active_with_window(self) -> None:
"""A configured window drives the curfew state."""
policy = _policy(curfew=CurfewWindow(start=time(23, 0), end=time(5, 0)))
assert policy.is_curfew_active(time(23, 30))
assert not policy.is_curfew_active(time(12, 0))
def test_packages_to_block_ignores_uninstalled(self) -> None:
"""Only packages present on the device can be blocked."""
policy = _policy()
installed = frozenset({"com.good", "com.evil", "com.android.settings"})
assert policy.packages_to_block(installed) == frozenset({"com.evil"})
def test_packages_to_block_during_curfew(self) -> None:
"""The curfew list is applied when the curfew flag is set."""
policy = _policy()
installed = frozenset({"com.good", "pl.mbank"})
assert policy.packages_to_block(installed, during_curfew=True) == frozenset(
{"com.good"},
)